MINOR: remove some specifying types in tool command

core/src/main/scala/kafka/admin/AclCommand.scala

@@ -319,7 +319,7 @@ object AclCommand extends Logging {
   }
 
   private def getResourceToAcls(opts: AclCommandOptions): Map[ResourcePattern, Set[AccessControlEntry]] = {
-    val patternType: PatternType = opts.options.valueOf(opts.resourcePatternType)
+    val patternType = opts.options.valueOf(opts.resourcePatternType)
     if (!patternType.isSpecific)
       CommandLineUtils.printUsageAndDie(opts.parser, s"A '--resource-pattern-type' value of '$patternType' is not valid when adding acls.")
 
@@ -357,8 +357,8 @@ object AclCommand extends Logging {
   private def getProducerResourceFilterToAcls(opts: AclCommandOptions): Map[ResourcePatternFilter, Set[AccessControlEntry]] = {
     val filters = getResourceFilter(opts)
 
-    val topics: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.TOPIC)
-    val transactionalIds: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.TRANSACTIONAL_ID)
+    val topics = filters.filter(_.resourceType == JResourceType.TOPIC)
+    val transactionalIds = filters.filter(_.resourceType == JResourceType.TRANSACTIONAL_ID)
     val enableIdempotence = opts.options.has(opts.idempotentOpt)
 
     val topicAcls = getAcl(opts, Set(WRITE, DESCRIBE, CREATE))
@@ -376,8 +376,8 @@ object AclCommand extends Logging {
   private def getConsumerResourceFilterToAcls(opts: AclCommandOptions): Map[ResourcePatternFilter, Set[AccessControlEntry]] = {
     val filters = getResourceFilter(opts)
 
-    val topics: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.TOPIC)
-    val groups: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.GROUP)
+    val topics = filters.filter(_.resourceType == JResourceType.TOPIC)
+    val groups = filters.filter(_.resourceType == JResourceType.GROUP)
 
     //Read, Describe on topic, Read on consumerGroup
 
@@ -445,7 +445,7 @@ object AclCommand extends Logging {
   }
 
   private def getResourceFilter(opts: AclCommandOptions, dieIfNoResourceFound: Boolean = true): Set[ResourcePatternFilter] = {
-    val patternType: PatternType = opts.options.valueOf(opts.resourcePatternType)
+    val patternType = opts.options.valueOf(opts.resourcePatternType)
 
     var resourceFilters = Set.empty[ResourcePatternFilter]
     if (opts.options.has(opts.topicOpt))

core/src/main/scala/kafka/admin/BrokerApiVersionsCommand.scala

@@ -108,7 +108,7 @@ object BrokerApiVersionsCommand {
                             val client: ConsumerNetworkClient,
                             val bootstrapBrokers: List[Node]) extends Logging {
 
-    @volatile var running: Boolean = true
+    @volatile var running = true
     val pendingFutures = new ConcurrentLinkedQueue[RequestFuture[ClientResponse]]()
 
     val networkThread = new KafkaThread("admin-client-network-thread", () => {
@@ -135,7 +135,7 @@ object BrokerApiVersionsCommand {
     private def send(target: Node,
                      api: ApiKeys,
                      request: AbstractRequest.Builder[_ <: AbstractRequest]): AbstractResponse = {
-      val future: RequestFuture[ClientResponse] = client.send(target, request)
+      val future = client.send(target, request)
       pendingFutures.add(future)
       future.awaitDone(Long.MaxValue, TimeUnit.MILLISECONDS)
       pendingFutures.remove(future)

core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala

@@ -21,7 +21,6 @@ import collection._
 import java.util.Properties
 import java.util.concurrent.ExecutionException
 
-import joptsimple.OptionSpecBuilder
 import kafka.common.AdminCommandFailedException
 import kafka.utils._
 import kafka.utils.Implicits._
@@ -129,11 +128,11 @@ object PreferredReplicaLeaderElectionCommand extends Logging {
       .describedAs("list of partitions for which preferred replica leader election needs to be triggered")
       .ofType(classOf[String])
 
-    private val zookeeperOptBuilder: OptionSpecBuilder = parser.accepts("zookeeper",
+    private val zookeeperOptBuilder = parser.accepts("zookeeper",
       "DEPRECATED. The connection string for the zookeeper connection in the " +
       "form host:port. Multiple URLS can be given to allow fail-over. " +
       "Replaced by --bootstrap-server, REQUIRED unless --bootstrap-server is given.")
-    private val bootstrapOptBuilder: OptionSpecBuilder = parser.accepts("bootstrap-server",
+    private val bootstrapOptBuilder = parser.accepts("bootstrap-server",
       "A hostname and port for the broker to connect to, " +
       "in the form host:port. Multiple comma-separated URLs can be given. REQUIRED unless --zookeeper is given.")
     parser.mutuallyExclusive(zookeeperOptBuilder, bootstrapOptBuilder)

core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala

@@ -443,7 +443,7 @@ object ReassignPartitionsCommand extends Logging {
     val (foundReassignments, notFoundReassignments) = targetReassignments.partition {
       case (part, _) => currentReassignments.contains(part)
     }
-    val foundResults: Seq[(TopicPartition, PartitionReassignmentState)] = foundReassignments.map {
+    val foundResults = foundReassignments.map {
       case (part, targetReplicas) => (part,
         PartitionReassignmentState(
           currentReassignments(part).replicas.
@@ -458,7 +458,7 @@ object ReassignPartitionsCommand extends Logging {
     }
     val topicDescriptions = adminClient.
       describeTopics(topicNamesToLookUp.asJava).values().asScala
-    val notFoundResults: Seq[(TopicPartition, PartitionReassignmentState)] = notFoundReassignments.map {
+    val notFoundResults = notFoundReassignments.map {
       case (part, targetReplicas) =>
         currentReassignments.get(part) match {
           case Some(reassignment) => (part,
@@ -557,7 +557,7 @@ object ReassignPartitionsCommand extends Logging {
     val replicaLogDirInfos = adminClient.describeReplicaLogDirs(
       targetMoves.keySet.asJava).all().get().asScala
     targetMoves.map { case (replica, targetLogDir) =>
-      val moveState: LogDirMoveState = replicaLogDirInfos.get(replica) match {
+      val moveState = replicaLogDirInfos.get(replica) match {
         case None => MissingReplicaMoveState(targetLogDir)
         case Some(info) => if (info.getCurrentReplicaLogDir == null) {
             MissingLogDirMoveState(targetLogDir)
@@ -1189,8 +1189,7 @@ object ReassignPartitionsCommand extends Logging {
    */
   def alterPartitionReassignments(adminClient: Admin,
                                   reassignments: Map[TopicPartition, Seq[Int]]): Map[TopicPartition, Throwable] = {
-    val results: Map[TopicPartition, KafkaFuture[Void]] =
-      adminClient.alterPartitionReassignments(reassignments.map { case (part, replicas) =>
+    val results = adminClient.alterPartitionReassignments(reassignments.map { case (part, replicas) =>
         (part, Optional.of(new NewPartitionReassignment(replicas.map(Integer.valueOf).asJava)))
       }.asJava).values().asScala
     results.flatMap {
@@ -1215,8 +1214,7 @@ object ReassignPartitionsCommand extends Logging {
   def cancelPartitionReassignments(adminClient: Admin,
                                    reassignments: Set[TopicPartition])
   : Map[TopicPartition, Throwable] = {
-    val results: Map[TopicPartition, KafkaFuture[Void]] =
-      adminClient.alterPartitionReassignments(reassignments.map {
+    val results = adminClient.alterPartitionReassignments(reassignments.map {
           (_, (None: Option[NewPartitionReassignment]).asJava)
         }.toMap.asJava).values().asScala
     results.flatMap { case (part, future) =>

core/src/main/scala/kafka/admin/TopicCommand.scala

@@ -91,13 +91,13 @@ object TopicCommand extends Logging {
   }
 
   class CommandTopicPartition(opts: TopicCommandOptions) {
-    val name: String = opts.topic.get
-    val partitions: Option[Integer] = opts.partitions
-    val replicationFactor: Option[Integer] = opts.replicationFactor
-    val replicaAssignment: Option[Map[Int, List[Int]]] = opts.replicaAssignment
-    val configsToAdd: Properties = parseTopicConfigsToBeAdded(opts)
-    val configsToDelete: Seq[String] = parseTopicConfigsToBeDeleted(opts)
-    val rackAwareMode: RackAwareMode = opts.rackAwareMode
+    val name = opts.topic.get
+    val partitions = opts.partitions
+    val replicationFactor = opts.replicationFactor
+    val replicaAssignment = opts.replicaAssignment
+    val configsToAdd = parseTopicConfigsToBeAdded(opts)
+    val configsToDelete = parseTopicConfigsToBeDeleted(opts)
+    val rackAwareMode = opts.rackAwareMode
 
     def hasReplicaAssignment: Boolean = replicaAssignment.isDefined
     def hasPartitions: Boolean = partitions.isDefined
@@ -170,12 +170,12 @@ object TopicCommand extends Logging {
   }
 
   class DescribeOptions(opts: TopicCommandOptions, liveBrokers: Set[Int]) {
-    val describeConfigs: Boolean =
+    val describeConfigs =
       !opts.reportUnavailablePartitions &&
       !opts.reportUnderReplicatedPartitions &&
       !opts.reportUnderMinIsrPartitions &&
       !opts.reportAtMinIsrPartitions
-    val describePartitions: Boolean = !opts.reportOverriddenConfigs
+    val describePartitions = !opts.reportOverriddenConfigs
 
     private def shouldPrintUnderReplicatedPartitions(partitionDescription: PartitionDescription): Boolean = {
       opts.reportUnderReplicatedPartitions && partitionDescription.isUnderReplicated
@@ -692,7 +692,7 @@ object TopicCommand extends Logging {
 
     private val allTopicLevelOpts = immutable.Set[OptionSpec[_]](alterOpt, createOpt, describeOpt, listOpt, deleteOpt)
 
-    private val allReplicationReportOpts: Set[OptionSpec[_]] = Set(reportUnderReplicatedPartitionsOpt, reportUnderMinIsrPartitionsOpt, reportAtMinIsrPartitionsOpt, reportUnavailablePartitionsOpt)
+    private val allReplicationReportOpts = Set(reportUnderReplicatedPartitionsOpt, reportUnderMinIsrPartitionsOpt, reportAtMinIsrPartitionsOpt, reportUnavailablePartitionsOpt)
 
     def has(builder: OptionSpec[_]): Boolean = options.has(builder)
     def valueAsOption[A](option: OptionSpec[A], defaultValue: Option[A] = None): Option[A] = if (has(option)) Some(options.valueOf(option)) else defaultValue

core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala

@@ -91,7 +91,7 @@ object ZkSecurityMigrator extends Logging {
       throw new IllegalArgumentException("Incorrect configuration")
     }
 
-    val zkAcl: Boolean = opts.options.valueOf(opts.zkAclOpt) match {
+    val zkAcl = opts.options.valueOf(opts.zkAclOpt) match {
       case "secure" =>
         info("zookeeper.acl option is secure")
         true