Skip to content

MINOR: remove some specifying types in tool command #10329

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chia7712 merged 6 commits into from
Mar 17, 2021
Merged

MINOR: remove some specifying types in tool command #10329

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
06dc4a0
MINOR: remove some specifying types in tool command
Mar 16, 2021
bed68ab
Merge remote-tracking branch 'sri/trunk' into removeSpecifyType
wenbingshen Mar 16, 2021
2d2d15f
MINOR: remove some specifying types in tool command
wenbingshen Mar 16, 2021
c599f56
Merge branch 'trunk' into removeSpecifyType
Mar 17, 2021
1eb3de6
MINOR: remove some specifying types in tool command
Mar 17, 2021
a31bf6f
MINOR: remove some specifying types in tool command
Mar 17, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions core/src/main/scala/kafka/admin/AclCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.")

Expand Down Expand Up @@ -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))
Expand All @@ -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

Expand Down Expand Up @@ -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))
Expand Down
4 changes: 2 additions & 2 deletions core/src/main/scala/kafka/admin/BrokerApiVersionsCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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", () => {
Expand All @@ -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)
Expand Down
5 changes: 2 additions & 3 deletions core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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._
Expand Down Expand Up @@ -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)
Expand Down
20 changes: 9 additions & 11 deletions core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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,
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -1189,10 +1189,9 @@ 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) =>
(part, Optional.of(new NewPartitionReassignment(replicas.map(Integer.valueOf).asJava)))
}.asJava).values().asScala
val results = adminClient.alterPartitionReassignments(reassignments.map { case (part, replicas) =>

@dajac dajac Mar 17, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: The indentation of the two lines below could be corrected now.

(part, Optional.of(new NewPartitionReassignment(replicas.map(Integer.valueOf).asJava)))
}.asJava).values().asScala
results.flatMap {
case (part, future) => {
try {
Expand All @@ -1215,10 +1214,9 @@ object ReassignPartitionsCommand extends Logging {
def cancelPartitionReassignments(adminClient: Admin,
reassignments: Set[TopicPartition])
: Map[TopicPartition, Throwable] = {
val results: Map[TopicPartition, KafkaFuture[Void]] =
adminClient.alterPartitionReassignments(reassignments.map {
(_, (None: Option[NewPartitionReassignment]).asJava)
}.toMap.asJava).values().asScala
val results = adminClient.alterPartitionReassignments(reassignments.map {

@dajac dajac Mar 17, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ditto

(_, (None: Option[NewPartitionReassignment]).asJava)

@dajac dajac Mar 17, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

While we are here, could we replace (None: Option[NewPartitionReassignment]).asJava by Optional.empty[NewPartitionReassignment]() or Option.empty[NewPartitionReassignment].asJava? Personally, I prefer using Optional directly here as we use Optional.of above.

@wenbingshen wenbingshen Mar 17, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good idea.I updated the code, please review it again, thank you.

}.toMap.asJava).values().asScala
results.flatMap { case (part, future) =>
try {
future.get()
Expand Down
20 changes: 10 additions & 10 deletions core/src/main/scala/kafka/admin/TopicCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down