Skip to content

MINOR; Various code cleanups #10319

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
dajac merged 3 commits into from
Mar 16, 2021
Merged

MINOR; Various code cleanups #10319

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f8eb1f8
MINOR; Various code cleanups
dajac Mar 15, 2021
205c17a
Address comments
dajac Mar 16, 2021
1630a8f
Address review + improve a try/catch
dajac Mar 16, 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
21 changes: 7 additions & 14 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 @@ -815,20 +815,13 @@ object ReassignPartitionsCommand extends Logging {
private def describeTopics(adminClient: Admin,
topics: Set[String])
: Map[String, TopicDescription] = {
adminClient.describeTopics(topics.asJava).values.asScala.map {
case (topicName, topicDescriptionFuture) =>
try {
topicName -> topicDescriptionFuture.get
}
catch {
case t: ExecutionException =>
if (classOf[UnknownTopicOrPartitionException].isInstance(t.getCause)) {
throw new ExecutionException(
new UnknownTopicOrPartitionException(s"Topic $topicName not found."))
} else {
throw t
}
}
adminClient.describeTopics(topics.asJava).values.asScala.map { case (topicName, topicDescriptionFuture) =>
try topicName -> topicDescriptionFuture.get
catch {
case t: ExecutionException if t.getCause.isInstanceOf[UnknownTopicOrPartitionException] =>
throw new ExecutionException(
new UnknownTopicOrPartitionException(s"Topic $topicName not found."))
}
}
}

Expand Down
2 changes: 1 addition & 1 deletion core/src/test/scala/unit/kafka/admin/DeleteTopicTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,7 +160,7 @@ class DeleteTopicTest extends ZooKeeperTestHarness {
partition: TopicPartition,
reassignment: NewPartitionReassignment): Unit = {
val e = assertThrows(classOf[ExecutionException], () => adminClient.alterPartitionReassignments(Collections.singletonMap(partition,
Optional.of(new NewPartitionReassignment(util.Arrays.asList(1, 2, 3))))).all().get())
Optional.of(reassignment))).all().get())
assertEquals(classOf[UnknownTopicOrPartitionException], e.getCause.getClass)
}

Expand Down
7 changes: 3 additions & 4 deletions core/src/test/scala/unit/kafka/admin/PreferredReplicaLeaderElectionCommandTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,11 @@ import java.nio.charset.StandardCharsets
import java.nio.file.{Files, Paths}
import java.util
import java.util.Properties

import scala.collection.Seq
import kafka.common.AdminCommandFailedException
import kafka.security.authorizer.AclAuthorizer
import kafka.server.{KafkaConfig, KafkaServer}
import kafka.utils.Implicits.MapExtensionMethods
import kafka.utils.{Logging, TestUtils}
import kafka.zk.ZooKeeperTestHarness
import org.apache.kafka.common.{TopicPartition, Uuid}
Expand Down Expand Up @@ -69,9 +69,8 @@ class PreferredReplicaLeaderElectionCommandTest extends ZooKeeperTestHarness wit
// create brokers
servers = brokerConfigs.map(b => TestUtils.createServer(KafkaConfig.fromProps(b)))
// create the topic
partitionsAndAssignments.foreach { case (tp, assignment) =>
zkClient.createTopicAssignment(tp.topic, Some(Uuid.randomUuid()),
Map(tp -> assignment))
partitionsAndAssignments.forKeyValue { (tp, assignment) =>
zkClient.createTopicAssignment(tp.topic, Some(Uuid.randomUuid()), Map(tp -> assignment))
}
// wait until replica log is created on every broker
TestUtils.waitUntilTrue(
Expand Down
5 changes: 3 additions & 2 deletions core/src/test/scala/unit/kafka/network/ConnectionQuotasTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import com.yammer.metrics.core.Meter
import kafka.metrics.KafkaMetricsGroup
import kafka.network.Processor.ListenerMetricTag
import kafka.server.KafkaConfig
import kafka.utils.Implicits.MapExtensionMethods
import kafka.utils.{MockTime, TestUtils}
import org.apache.kafka.common.config.ConfigException
import org.apache.kafka.common.config.internals.QuotaConfigs
Expand Down Expand Up @@ -782,14 +783,14 @@ class ConnectionQuotasTest {
}

private def verifyNoBlockedPercentRecordedOnAllListeners(): Unit = {
blockedPercentMeters.foreach { case (name, meter) =>
blockedPercentMeters.forKeyValue { (name, meter) =>

@chia7712 chia7712 Mar 16, 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.

line#809 and line#761 have similar issue. Could you fix them in this PR?

assertEquals(0, meter.count(),
s"BlockedPercentMeter metric for $name listener")
}
}

private def verifyNonZeroBlockedPercentAndThrottleTimeOnAllListeners(): Unit = {
blockedPercentMeters.foreach { case (name, meter) =>
blockedPercentMeters.forKeyValue { (name, meter) =>
assertTrue(meter.count() > 0,
s"Expected BlockedPercentMeter metric for $name listener to be recorded")
}
Expand Down