You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2022/11/25 10:14:30 UTC
[kafka] branch trunk updated: MINOR: Improve Scala collection usage (#12900)
This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new c8a228d122b MINOR: Improve Scala collection usage (#12900)
c8a228d122b is described below
commit c8a228d122b32d9df9f56b43fe55a27a07ee6a12
Author: Divij Vaidya <di...@amazon.com>
AuthorDate: Fri Nov 25 11:14:21 2022 +0100
MINOR: Improve Scala collection usage (#12900)
Reviewers: Mickael Maison <mi...@gmail.com>
, Christo Lolov <ch...@yahoo.com>
---
core/src/main/scala/kafka/Kafka.scala | 2 +-
core/src/main/scala/kafka/admin/ConfigCommand.scala | 2 +-
core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala | 6 +++---
core/src/main/scala/kafka/admin/TopicCommand.scala | 4 ++--
core/src/main/scala/kafka/cluster/Partition.scala | 2 +-
core/src/main/scala/kafka/controller/ControllerChannelManager.scala | 2 +-
core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala | 2 +-
core/src/main/scala/kafka/metrics/KafkaMetricsGroup.scala | 2 +-
core/src/main/scala/kafka/server/DynamicConfig.scala | 2 +-
core/src/main/scala/kafka/server/KafkaConfig.scala | 6 +++---
core/src/main/scala/kafka/server/ZkAdminManager.scala | 6 +++---
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala | 4 ++--
core/src/main/scala/kafka/tools/GetOffsetShell.scala | 2 +-
core/src/main/scala/kafka/tools/JmxTool.scala | 2 +-
core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala | 2 +-
core/src/main/scala/kafka/tools/StateChangeLogMerger.scala | 2 +-
core/src/main/scala/kafka/utils/CommandLineUtils.scala | 4 ++--
core/src/main/scala/kafka/zk/KafkaZkClient.scala | 4 ++--
core/src/test/scala/kafka/tools/LogCompactionTester.scala | 2 +-
core/src/test/scala/unit/kafka/admin/AclCommandTest.scala | 2 +-
.../src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala | 6 +++---
core/src/test/scala/unit/kafka/cluster/PartitionTest.scala | 4 ++--
core/src/test/scala/unit/kafka/log/LogCleanerTest.scala | 4 ++--
core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala | 2 +-
.../src/test/scala/unit/kafka/server/KafkaMetricsReporterTest.scala | 2 +-
core/src/test/scala/unit/kafka/utils/TestUtils.scala | 2 +-
26 files changed, 40 insertions(+), 40 deletions(-)
diff --git a/core/src/main/scala/kafka/Kafka.scala b/core/src/main/scala/kafka/Kafka.scala
index 5dc829fd3b1..e1bd575f0c2 100755
--- a/core/src/main/scala/kafka/Kafka.scala
+++ b/core/src/main/scala/kafka/Kafka.scala
@@ -40,7 +40,7 @@ object Kafka extends Logging {
// This is a bit of an ugly crutch till we get a chance to rework the entire command line parsing
optionParser.accepts("version", "Print version information and exit.")
- if (args.length == 0 || args.contains("--help")) {
+ if (args.isEmpty || args.contains("--help")) {
CommandLineUtils.printUsageAndDie(optionParser,
"USAGE: java [options] %s server.properties [--override property=value]*".format(this.getClass.getCanonicalName.split('$').head))
}
diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index 9a42f9b874d..c5a62e7eb95 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -213,7 +213,7 @@ object ConfigCommand extends Logging {
throw new IllegalArgumentException("Password encoder secret not specified"))
PasswordEncoder.encrypting(new Password(encoderSecret),
None,
- encoderConfigs.get(KafkaConfig.PasswordEncoderCipherAlgorithmProp).getOrElse(Defaults.PasswordEncoderCipherAlgorithm),
+ encoderConfigs.getOrElse(KafkaConfig.PasswordEncoderCipherAlgorithmProp, Defaults.PasswordEncoderCipherAlgorithm),
encoderConfigs.get(KafkaConfig.PasswordEncoderKeyLengthProp).map(_.toInt).getOrElse(Defaults.PasswordEncoderKeyLength),
encoderConfigs.get(KafkaConfig.PasswordEncoderIterationsProp).map(_.toInt).getOrElse(Defaults.PasswordEncoderIterations))
}
diff --git a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
index 8cc056f617e..f3b67a030f9 100755
--- a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
@@ -298,7 +298,7 @@ object ConsumerGroupCommand extends Logging {
maxGroupInstanceIdLen = Math.max(maxGroupInstanceIdLen, memberAssignment.groupInstanceId.length)
maxHostLen = Math.max(maxHostLen, memberAssignment.host.length)
maxClientIdLen = Math.max(maxClientIdLen, memberAssignment.clientId.length)
- includeGroupInstanceId = includeGroupInstanceId || memberAssignment.groupInstanceId.length > 0
+ includeGroupInstanceId = includeGroupInstanceId || memberAssignment.groupInstanceId.nonEmpty
}
}
@@ -563,8 +563,8 @@ object ConsumerGroupCommand extends Logging {
// The admin client returns `null` as a value to indicate that there is not committed offset for a partition.
def getPartitionOffset(tp: TopicPartition): Option[Long] = committedOffsets.get(tp).filter(_ != null).map(_.offset)
var assignedTopicPartitions = ListBuffer[TopicPartition]()
- val rowsWithConsumer = consumerGroup.members.asScala.filter(!_.assignment.topicPartitions.isEmpty).toSeq
- .sortWith(_.assignment.topicPartitions.size > _.assignment.topicPartitions.size).flatMap { consumerSummary =>
+ val rowsWithConsumer = consumerGroup.members.asScala.filterNot(_.assignment.topicPartitions.isEmpty).toSeq
+ .sortBy(_.assignment.topicPartitions.size)(Ordering[Int].reverse).flatMap { consumerSummary =>
val topicPartitions = consumerSummary.assignment.topicPartitions.asScala
assignedTopicPartitions = assignedTopicPartitions ++ topicPartitions
collectConsumerAssignment(groupId, Option(consumerGroup.coordinator), topicPartitions.toList,
diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala
index 5e7d98c1d2e..e5d60670892 100755
--- a/core/src/main/scala/kafka/admin/TopicCommand.scala
+++ b/core/src/main/scala/kafka/admin/TopicCommand.scala
@@ -104,7 +104,7 @@ object TopicCommand extends Logging {
markedForDeletion: Boolean) {
def printDescription(): Unit = {
- val configsAsString = config.entries.asScala.filter(!_.isDefault).map { ce => s"${ce.name}=${ce.value}" }.mkString(",")
+ val configsAsString = config.entries.asScala.filterNot(_.isDefault).map { ce => s"${ce.name}=${ce.value}" }.mkString(",")
print(s"Topic: $topic")
if(topicId != Uuid.ZERO_UUID) print(s"\tTopicId: $topicId")
print(s"\tPartitionCount: $numPartitions")
@@ -604,7 +604,7 @@ object TopicCommand extends Logging {
def configsToDelete: Option[util.List[String]] = valuesAsOption(deleteConfigOpt)
def checkArgs(): Unit = {
- if (args.length == 0)
+ if (args.isEmpty)
CommandLineUtils.printUsageAndDie(parser, "Create, delete, describe, or change a topic.")
CommandLineUtils.printHelpAndExitIfNeeded(this, "This tool helps to create, delete, describe, or change a topic.")
diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala
index e1d2d855783..63f8f908859 100755
--- a/core/src/main/scala/kafka/cluster/Partition.scala
+++ b/core/src/main/scala/kafka/cluster/Partition.scala
@@ -808,7 +808,7 @@ class Partition(val topicPartition: TopicPartition,
): Unit = {
if (isLeader) {
val followers = replicas.filter(_ != localBrokerId)
- val removedReplicas = remoteReplicasMap.keys.filter(!followers.contains(_))
+ val removedReplicas = remoteReplicasMap.keys.filterNot(followers.contains(_))
// Due to code paths accessing remoteReplicasMap without a lock,
// first add the new replicas and then remove the old ones
diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
index d900a7ccea9..f6b9cbd07dd 100755
--- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
+++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
@@ -432,7 +432,7 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
controllerContext.partitionLeadershipInfo(partition) match {
case Some(LeaderIsrAndControllerEpoch(leaderAndIsr, controllerEpoch)) =>
val replicas = controllerContext.partitionReplicaAssignment(partition)
- val offlineReplicas = replicas.filter(!controllerContext.isReplicaOnline(_, partition))
+ val offlineReplicas = replicas.filterNot(controllerContext.isReplicaOnline(_, partition))
val updatedLeaderAndIsr =
if (beingDeleted) LeaderAndIsr.duringDelete(leaderAndIsr.isr)
else leaderAndIsr
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala
index c2e2176856c..3c146dd9c13 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala
@@ -384,7 +384,7 @@ private[group] class GroupMetadata(val groupId: String, initialState: GroupState
def currentState: GroupState = state
- def notYetRejoinedMembers: Map[String, MemberMetadata] = members.filter(!_._2.isAwaitingJoin).toMap
+ def notYetRejoinedMembers: Map[String, MemberMetadata] = members.filterNot(_._2.isAwaitingJoin).toMap
def hasAllMembersJoined: Boolean = members.size == numMembersAwaitingJoin && pendingMembers.isEmpty
diff --git a/core/src/main/scala/kafka/metrics/KafkaMetricsGroup.scala b/core/src/main/scala/kafka/metrics/KafkaMetricsGroup.scala
index 161d1f2f364..8a9ed89f3fc 100644
--- a/core/src/main/scala/kafka/metrics/KafkaMetricsGroup.scala
+++ b/core/src/main/scala/kafka/metrics/KafkaMetricsGroup.scala
@@ -97,7 +97,7 @@ trait KafkaMetricsGroup extends Logging {
if (filteredTags.nonEmpty) {
// convert dot to _ since reporters like Graphite typically use dot to represent hierarchy
val tagsString = filteredTags
- .toList.sortWith((t1, t2) => t1._1 < t2._1)
+ .toList.sortBy(_._1)
.map { case (key, value) => "%s.%s".format(key, value.replaceAll("\\.", "_"))}
.mkString(".")
diff --git a/core/src/main/scala/kafka/server/DynamicConfig.scala b/core/src/main/scala/kafka/server/DynamicConfig.scala
index ddcda030ad6..e3479e3641a 100644
--- a/core/src/main/scala/kafka/server/DynamicConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfig.scala
@@ -117,7 +117,7 @@ object DynamicConfig {
val names = configDef.names()
val propKeys = props.keySet.asScala.map(_.asInstanceOf[String])
if (!customPropsAllowed) {
- val unknownKeys = propKeys.filter(!names.contains(_))
+ val unknownKeys = propKeys.filterNot(names.contains(_))
require(unknownKeys.isEmpty, s"Unknown Dynamic Configuration: $unknownKeys.")
}
val propResolved = DynamicBrokerConfig.resolveVariableConfigs(props)
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 2f9e9e6dba0..98953e338de 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -667,7 +667,7 @@ object KafkaConfig {
val ZkSslClientEnableDoc = "Set client to use TLS when connecting to ZooKeeper." +
" An explicit value overrides any value set via the <code>zookeeper.client.secure</code> system property (note the different name)." +
s" Defaults to false if neither is set; when true, <code>$ZkClientCnxnSocketProp</code> must be set (typically to <code>org.apache.zookeeper.ClientCnxnSocketNetty</code>); other values to set may include " +
- ZkSslConfigToSystemPropertyMap.keys.toList.sorted.filter(x => x != ZkSslClientEnableProp && x != ZkClientCnxnSocketProp).mkString("<code>", "</code>, <code>", "</code>")
+ ZkSslConfigToSystemPropertyMap.keys.toList.filter(x => x != ZkSslClientEnableProp && x != ZkClientCnxnSocketProp).sorted.mkString("<code>", "</code>, <code>", "</code>")
val ZkClientCnxnSocketDoc = "Typically set to <code>org.apache.zookeeper.ClientCnxnSocketNetty</code> when using TLS connectivity to ZooKeeper." +
s" Overrides any explicit value set via the same-named <code>${ZkSslConfigToSystemPropertyMap(ZkClientCnxnSocketProp)}</code> system property."
val ZkSslKeyStoreLocationDoc = "Keystore location when using a client-side certificate with TLS connectivity to ZooKeeper." +
@@ -1732,7 +1732,7 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
Option(getString(KafkaConfig.EarlyStartListenersProp)) match {
case None => controllerListenersSet
case Some(str) =>
- str.split(",").map(_.trim()).filter(!_.isEmpty).map { str =>
+ str.split(",").map(_.trim()).filterNot(_.isEmpty).map { str =>
val listenerName = new ListenerName(str)
if (!listenersSet.contains(listenerName) && !controllerListenersSet.contains(listenerName))
throw new ConfigException(s"${KafkaConfig.EarlyStartListenersProp} contains " +
@@ -2086,7 +2086,7 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
mapValue // don't add default mappings since we found something that is SSL or SASL_*
} else {
// add the PLAINTEXT mappings for all controller listener names that are not explicitly PLAINTEXT
- mapValue ++ controllerListenerNames.filter(!SecurityProtocol.PLAINTEXT.name.equals(_)).map(
+ mapValue ++ controllerListenerNames.filterNot(SecurityProtocol.PLAINTEXT.name.equals(_)).map(
new ListenerName(_) -> SecurityProtocol.PLAINTEXT)
}
} else {
diff --git a/core/src/main/scala/kafka/server/ZkAdminManager.scala b/core/src/main/scala/kafka/server/ZkAdminManager.scala
index f65367606da..634ce6b097c 100644
--- a/core/src/main/scala/kafka/server/ZkAdminManager.scala
+++ b/core/src/main/scala/kafka/server/ZkAdminManager.scala
@@ -493,7 +493,7 @@ class ZkAdminManager(val config: KafkaConfig,
resource.`type` match {
case ConfigResource.Type.TOPIC =>
- if (resource.name.isEmpty()) {
+ if (resource.name.isEmpty) {
throw new InvalidRequestException("Default topic resources are not allowed.")
}
val configProps = adminZkClient.fetchEntityConfig(ConfigType.Topic, resource.name)
@@ -696,11 +696,11 @@ class ZkAdminManager(val config: KafkaConfig,
def matches(nameComponent: Option[ClientQuotaFilterComponent], name: Option[String]): Boolean = nameComponent match {
case Some(component) =>
toOption(component.`match`) match {
- case Some(n) => name.exists(_ == n)
+ case Some(n) => name.contains(n)
case None => name.isDefined
}
case None =>
- !name.isDefined || !strict
+ name.isEmpty || !strict
}
(userEntries ++ clientIdEntries ++ bothEntries).flatMap { case ((u, c), p) =>
diff --git a/core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala b/core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala
index 52577211503..7bd1c6343d9 100644
--- a/core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala
+++ b/core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala
@@ -213,7 +213,7 @@ class KRaftMetadataCache(val brokerId: Int) extends MetadataCache with Logging w
override def getAliveBrokers(): Iterable[BrokerMetadata] = getAliveBrokers(_currentImage)
private def getAliveBrokers(image: MetadataImage): Iterable[BrokerMetadata] = {
- image.cluster().brokers().values().asScala.filter(!_.fenced()).
+ image.cluster().brokers().values().asScala.filterNot(_.fenced()).
map(b => BrokerMetadata(b.id, b.rack.asScala))
}
@@ -223,7 +223,7 @@ class KRaftMetadataCache(val brokerId: Int) extends MetadataCache with Logging w
}
override def getAliveBrokerNodes(listenerName: ListenerName): Seq[Node] = {
- _currentImage.cluster().brokers().values().asScala.filter(!_.fenced()).
+ _currentImage.cluster().brokers().values().asScala.filterNot(_.fenced()).
flatMap(_.node(listenerName.value()).asScala).toSeq
}
diff --git a/core/src/main/scala/kafka/tools/GetOffsetShell.scala b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
index 03f9c819260..a8fd87cbe87 100644
--- a/core/src/main/scala/kafka/tools/GetOffsetShell.scala
+++ b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
@@ -81,7 +81,7 @@ object GetOffsetShell {
.ofType(classOf[String])
val excludeInternalTopicsOpt = parser.accepts("exclude-internal-topics", s"By default, internal topics are included. If specified, internal topics are excluded.")
- if (args.length == 0)
+ if (args.isEmpty)
CommandLineUtils.printUsageAndDie(parser, "An interactive shell for getting topic-partition offsets.")
val options = parser.parse(args : _*)
diff --git a/core/src/main/scala/kafka/tools/JmxTool.scala b/core/src/main/scala/kafka/tools/JmxTool.scala
index f7ace833728..c0f6d4a5ead 100644
--- a/core/src/main/scala/kafka/tools/JmxTool.scala
+++ b/core/src/main/scala/kafka/tools/JmxTool.scala
@@ -98,7 +98,7 @@ object JmxTool extends Logging {
val helpOpt = parser.accepts("help", "Print usage information.")
- if(args.length == 0)
+ if(args.isEmpty)
CommandLineUtils.printUsageAndDie(parser, "Dump JMX values to standard output.")
val options = parser.parse(args : _*)
diff --git a/core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala b/core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala
index a9bd2c936e1..fb687e4f8e8 100644
--- a/core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala
+++ b/core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala
@@ -114,7 +114,7 @@ object ReplicaVerificationTool extends Logging {
val options = parser.parse(args: _*)
- if (args.length == 0 || options.has(helpOpt)) {
+ if (args.isEmpty || options.has(helpOpt)) {
CommandLineUtils.printUsageAndDie(parser, "Validate that all replicas for a set of topics have the same data.")
}
diff --git a/core/src/main/scala/kafka/tools/StateChangeLogMerger.scala b/core/src/main/scala/kafka/tools/StateChangeLogMerger.scala
index 1a7f679fe36..32e43a6606d 100755
--- a/core/src/main/scala/kafka/tools/StateChangeLogMerger.scala
+++ b/core/src/main/scala/kafka/tools/StateChangeLogMerger.scala
@@ -88,7 +88,7 @@ object StateChangeLogMerger extends Logging {
.ofType(classOf[String])
.defaultsTo("9999-12-31 23:59:59,999")
- if(args.length == 0)
+ if(args.isEmpty)
CommandLineUtils.printUsageAndDie(parser, "A tool for merging the log files from several brokers to reconnstruct a unified history of what happened.")
diff --git a/core/src/main/scala/kafka/utils/CommandLineUtils.scala b/core/src/main/scala/kafka/utils/CommandLineUtils.scala
index 1eaee484168..0aa233babcd 100644
--- a/core/src/main/scala/kafka/utils/CommandLineUtils.scala
+++ b/core/src/main/scala/kafka/utils/CommandLineUtils.scala
@@ -33,7 +33,7 @@ object CommandLineUtils extends Logging {
* @return true on matching the help check condition
*/
def isPrintHelpNeeded(commandOpts: CommandDefaultOptions): Boolean = {
- commandOpts.args.length == 0 || commandOpts.options.has(commandOpts.helpOpt)
+ commandOpts.args.isEmpty || commandOpts.options.has(commandOpts.helpOpt)
}
def isPrintVersionNeeded(commandOpts: CommandDefaultOptions): Boolean = {
@@ -113,7 +113,7 @@ object CommandLineUtils extends Logging {
* value may contain equals sign
*/
def parseKeyValueArgs(args: Iterable[String], acceptMissingValue: Boolean = true): Properties = {
- val splits = args.map(_.split("=", 2)).filterNot(_.length == 0)
+ val splits = args.map(_.split("=", 2)).filterNot(_.isEmpty)
val props = new Properties
for (a <- splits) {
diff --git a/core/src/main/scala/kafka/zk/KafkaZkClient.scala b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
index fa7ce00882a..747673d37db 100644
--- a/core/src/main/scala/kafka/zk/KafkaZkClient.scala
+++ b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
@@ -211,7 +211,7 @@ class KafkaZkClient private[zk] (zooKeeperClient: ZooKeeperClient, isSecure: Boo
* @return sequence of CreateResponse whose contexts are the partitions they are associated with.
*/
def createTopicPartitionStatesRaw(leaderIsrAndControllerEpochs: Map[TopicPartition, LeaderIsrAndControllerEpoch], expectedControllerEpochZkVersion: Int): Seq[CreateResponse] = {
- createTopicPartitions(leaderIsrAndControllerEpochs.keys.map(_.topic).toSet.toSeq, expectedControllerEpochZkVersion)
+ createTopicPartitions(leaderIsrAndControllerEpochs.keys.map(_.topic).toSeq.distinct, expectedControllerEpochZkVersion)
createTopicPartition(leaderIsrAndControllerEpochs.keys.toSeq, expectedControllerEpochZkVersion)
val createRequests = leaderIsrAndControllerEpochs.map { case (partition, leaderIsrAndControllerEpoch) =>
val path = TopicPartitionStateZNode.path(partition)
@@ -701,7 +701,7 @@ class KafkaZkClient private[zk] (zooKeeperClient: ZooKeeperClient, isSecure: Boo
getPartitionAssignmentForTopics(topics).map { topicAndPartitionMap =>
val topic = topicAndPartitionMap._1
val partitionMap = topicAndPartitionMap._2
- topic -> partitionMap.keys.toSeq.sortWith((s, t) => s < t)
+ topic -> partitionMap.keys.toSeq.sorted
}
}
diff --git a/core/src/test/scala/kafka/tools/LogCompactionTester.scala b/core/src/test/scala/kafka/tools/LogCompactionTester.scala
index da8a3c03a17..1b3b9f5b18f 100755
--- a/core/src/test/scala/kafka/tools/LogCompactionTester.scala
+++ b/core/src/test/scala/kafka/tools/LogCompactionTester.scala
@@ -97,7 +97,7 @@ object LogCompactionTester {
val options = parser.parse(args: _*)
- if (args.length == 0)
+ if (args.isEmpty)
CommandLineUtils.printUsageAndDie(parser, "A tool to test log compaction. Valid options are: ")
CommandLineUtils.checkRequiredArgs(parser, options, brokerOpt, numMessagesOpt)
diff --git a/core/src/test/scala/unit/kafka/admin/AclCommandTest.scala b/core/src/test/scala/unit/kafka/admin/AclCommandTest.scala
index b5d9692040e..dd81ede4604 100644
--- a/core/src/test/scala/unit/kafka/admin/AclCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/AclCommandTest.scala
@@ -175,7 +175,7 @@ class AclCommandTest extends QuorumTestHarness with Logging {
private def assertOutputContains(prefix: String, resources: Set[ResourcePattern], resourceCmd: Array[String], output: String): Unit = {
resources.foreach { resource =>
val resourceType = resource.resourceType.toString
- (if (resource == ClusterResource) Array("kafka-cluster") else resourceCmd.filter(!_.startsWith("--"))).foreach { name =>
+ (if (resource == ClusterResource) Array("kafka-cluster") else resourceCmd.filterNot(_.startsWith("--"))).foreach { name =>
val expected = s"$prefix for resource `ResourcePattern(resourceType=$resourceType, name=$name, patternType=LITERAL)`:"
assertTrue(output.contains(expected), s"Substring $expected not in output:\n$output")
}
diff --git a/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala b/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala
index 3815705971c..d8127746163 100644
--- a/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala
@@ -184,7 +184,7 @@ class DescribeConsumerGroupTest extends ConsumerGroupCommandTest {
TestUtils.waitUntilTrue(() => {
val (output, error) = TestUtils.grabConsoleOutputAndError(service.describeGroups())
- val numLines = output.trim.split("\n").filterNot(line => line.isEmpty).length
+ val numLines = output.trim.split("\n").count(line => line.nonEmpty)
(numLines == expectedNumLines) && error.isEmpty
}, s"Expected a data row and no error in describe results with describe type ${describeType.mkString(" ")}.")
}
@@ -208,7 +208,7 @@ class DescribeConsumerGroupTest extends ConsumerGroupCommandTest {
TestUtils.waitUntilTrue(() => {
val (output, error) = TestUtils.grabConsoleOutputAndError(service.describeGroups())
- val numLines = output.trim.split("\n").filterNot(line => line.isEmpty).length
+ val numLines = output.trim.split("\n").count(line => line.nonEmpty)
(numLines == expectedNumLines) && error.isEmpty
}, s"Expected a data row and no error in describe results with describe type ${describeType.mkString(" ")}.")
}
@@ -468,7 +468,7 @@ class DescribeConsumerGroupTest extends ConsumerGroupCommandTest {
assignments.get.count(_.group == group) == 2 &&
assignments.get.count { x => x.group == group && x.numPartitions == 1 } == 1 &&
assignments.get.count { x => x.group == group && x.numPartitions == 0 } == 1 &&
- assignments.get.count(_.assignment.nonEmpty) == 0
+ !assignments.get.exists(_.assignment.nonEmpty)
}, "Expected rows for consumers with no assigned partitions in describe group results")
val (state, assignments) = service.collectGroupMembers(group, true)
diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
index 7abff0dc3ff..0c5453b54d3 100644
--- a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
+++ b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
@@ -2334,9 +2334,9 @@ class PartitionTest extends AbstractPartitionTest {
"AtMinIsr")
def getMetric(metric: String): Option[Metric] = {
- KafkaYammerMetrics.defaultRegistry().allMetrics().asScala.filter { case (metricName, _) =>
+ KafkaYammerMetrics.defaultRegistry().allMetrics().asScala.find { case (metricName, _) =>
metricName.getName == metric && metricName.getType == "Partition"
- }.headOption.map(_._2)
+ }.map(_._2)
}
assertTrue(metricsToCheck.forall(getMetric(_).isDefined))
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
index 34963a41e06..819278bd280 100755
--- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
@@ -86,7 +86,7 @@ class LogCleanerTest {
val segments = log.logSegments.take(3).toSeq
val stats = new CleanerStats()
val expectedBytesRead = segments.map(_.size).sum
- val shouldRemain = LogTestUtils.keysInLog(log).filter(!keys.contains(_))
+ val shouldRemain = LogTestUtils.keysInLog(log).filterNot(keys.contains)
cleaner.cleanSegments(log, segments, map, 0L, stats, new CleanedTransactionMetadata, -1)
assertEquals(shouldRemain, LogTestUtils.keysInLog(log))
assertEquals(expectedBytesRead, stats.bytesRead)
@@ -844,7 +844,7 @@ class LogCleanerTest {
// clean the log
val stats = new CleanerStats()
cleaner.cleanSegments(log, Seq(log.logSegments.head), map, 0L, stats, new CleanedTransactionMetadata, -1)
- val shouldRemain = LogTestUtils.keysInLog(log).filter(!keys.contains(_))
+ val shouldRemain = LogTestUtils.keysInLog(log).filterNot(keys.contains)
assertEquals(shouldRemain, LogTestUtils.keysInLog(log))
}
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
index 904fbbc2165..4ff43cfc6af 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
@@ -588,7 +588,7 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
private def alterClientQuotas(request: Map[ClientQuotaEntity, Map[String, Option[Double]]], validateOnly: Boolean) = {
val entries = request.map { case (entity, alter) =>
val ops = alter.map { case (key, value) =>
- new ClientQuotaAlteration.Op(key, value.map(Double.box).getOrElse(null))
+ new ClientQuotaAlteration.Op(key, value.map(Double.box).orNull)
}.asJavaCollection
new ClientQuotaAlteration(entity, ops)
}
diff --git a/core/src/test/scala/unit/kafka/server/KafkaMetricsReporterTest.scala b/core/src/test/scala/unit/kafka/server/KafkaMetricsReporterTest.scala
index 40b839fdd38..db13f225359 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaMetricsReporterTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaMetricsReporterTest.scala
@@ -48,7 +48,7 @@ object KafkaMetricsReporterTest {
}
private def contextLabelOrNull(name: String, metricsContext: MetricsContext): String = {
- Option(metricsContext.contextLabels().get(name)).flatMap(v => Option(v.toString())).getOrElse(null)
+ Option(metricsContext.contextLabels().get(name)).flatMap(v => Option(v.toString())).orNull
}
override def configure(configs: util.Map[String, _]): Unit = {}
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index 882f64b569c..0162ebdd5e3 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -1895,7 +1895,7 @@ object TestUtils extends Logging {
def alterClientQuotas(adminClient: Admin, request: Map[ClientQuotaEntity, Map[String, Option[Double]]]): AlterClientQuotasResult = {
val entries = request.map { case (entity, alter) =>
val ops = alter.map { case (key, value) =>
- new ClientQuotaAlteration.Op(key, value.map(Double.box).getOrElse(null))
+ new ClientQuotaAlteration.Op(key, value.map(Double.box).orNull)
}.asJavaCollection
new ClientQuotaAlteration(entity, ops)
}.asJavaCollection