KAFKA-16610 Replace "Map#entrySet#forEach" by "Map#forEach" (#15795)

Reviewers: Apoorv Mittal <[email protected]>, Igor Soarez <[email protected]>
apache · Apr 25, 2024 · 864744f · 864744f
1 parent a7ceacd
commit 864744f
Show file tree

Hide file tree

Showing 12 changed files with 58 additions and 53 deletions.
diff --git a/core/src/main/scala/kafka/server/ControllerApis.scala b/core/src/main/scala/kafka/server/ControllerApis.scala
@@ -530,12 +530,12 @@ class ControllerApis(
         if (exception != null) {
           requestHelper.handleError(request, exception)
         } else {
-          controllerResults.entrySet().forEach(entry => response.responses().add(
+          controllerResults.forEach((key, value) => response.responses().add(
             new OldAlterConfigsResourceResponse().
-              setErrorCode(entry.getValue.error().code()).
-              setErrorMessage(entry.getValue.message()).
-              setResourceName(entry.getKey.name()).
-              setResourceType(entry.getKey.`type`().id())))
+              setErrorCode(value.error().code()).
+              setErrorMessage(value.message()).
+              setResourceName(key.name()).
+              setResourceType(key.`type`().id())))
           requestHelper.sendResponseMaybeThrottle(request, throttleMs =>
             new AlterConfigsResponse(response.setThrottleTimeMs(throttleMs)))
         }
@@ -771,12 +771,12 @@ class ControllerApis(
         if (exception != null) {
           requestHelper.handleError(request, exception)
         } else {
-          controllerResults.entrySet().forEach(entry => response.responses().add(
+          controllerResults.forEach((key, value) => response.responses().add(
             new AlterConfigsResourceResponse().
-              setErrorCode(entry.getValue.error().code()).
-              setErrorMessage(entry.getValue.message()).
-              setResourceName(entry.getKey.name()).
-              setResourceType(entry.getKey.`type`().id())))
+              setErrorCode(value.error().code()).
+              setErrorMessage(value.message()).
+              setResourceName(key.name()).
+              setResourceType(key.`type`().id())))
           brokerLoggerResponses.forEach(r => response.responses().add(r))
           requestHelper.sendResponseMaybeThrottle(request, throttleMs =>
             new IncrementalAlterConfigsResponse(response.setThrottleTimeMs(throttleMs)))

diff --git a/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala b/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala
@@ -96,11 +96,11 @@ class ControllerConfigurationValidator(kafkaConfig: KafkaConfig) extends Configu
         validateTopicName(resource.name())
         val properties = new Properties()
         val nullTopicConfigs = new mutable.ArrayBuffer[String]()
-        config.entrySet().forEach(e => {
-          if (e.getValue == null) {
-            nullTopicConfigs += e.getKey
+        config.forEach((key, value) => {
+          if (value == null) {
+            nullTopicConfigs += key
           } else {
-            properties.setProperty(e.getKey, e.getValue)
+            properties.setProperty(key, value)
           }
         })
         if (nullTopicConfigs.nonEmpty) {
@@ -111,7 +111,7 @@ class ControllerConfigurationValidator(kafkaConfig: KafkaConfig) extends Configu
       case BROKER => validateBrokerName(resource.name())
       case CLIENT_METRICS =>
         val properties = new Properties()
-        config.entrySet().forEach(e => properties.setProperty(e.getKey, e.getValue))
+        config.forEach((key, value) => properties.setProperty(key, value))
         ClientMetricsConfigs.validate(resource.name(), properties)
       case _ => throwExceptionForUnknownResourceType(resource)
     }

diff --git a/core/src/main/scala/kafka/server/metadata/AclPublisher.scala b/core/src/main/scala/kafka/server/metadata/AclPublisher.scala
@@ -70,11 +70,11 @@ class AclPublisher(
           try {
             // Because the changes map is a LinkedHashMap, the deltas will be returned in
             // the order they were performed.
-            aclsDelta.changes().entrySet().forEach(e =>
-              if (e.getValue.isPresent) {
-                authorizer.addAcl(e.getKey, e.getValue.get())
+            aclsDelta.changes().forEach((key, value) =>
+              if (value.isPresent) {
+                authorizer.addAcl(key, value.get())
               } else {
-                authorizer.removeAcl(e.getKey)
+                authorizer.removeAcl(key)
               })
           } catch {
             case t: Throwable => faultHandler.handleFault("Error loading " +

diff --git a/core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala b/core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala
@@ -51,8 +51,8 @@ class ClientQuotaMetadataManager(private[metadata] val quotaManagers: QuotaManag
                                  private[metadata] val connectionQuotas: ConnectionQuotas) extends Logging {
 
   def update(quotasDelta: ClientQuotasDelta): Unit = {
-    quotasDelta.changes().entrySet().forEach { e =>
-      update(e.getKey, e.getValue)
+    quotasDelta.changes().forEach { (key, value) =>
+      update(key, value)
     }
   }
 
@@ -96,8 +96,8 @@ class ClientQuotaMetadataManager(private[metadata] val quotaManagers: QuotaManag
           ClientIdEntity(clientIdVal)
         }
       }
-      quotaDelta.changes().entrySet().forEach { e =>
-        handleUserClientQuotaChange(userClientEntity, e.getKey, e.getValue.asScala)
+      quotaDelta.changes().forEach { (key, value) =>
+        handleUserClientQuotaChange(userClientEntity, key, value.asScala)
       }
     } else {
       warn(s"Ignoring unsupported quota entity $entity.")
@@ -116,10 +116,10 @@ class ClientQuotaMetadataManager(private[metadata] val quotaManagers: QuotaManag
       case _ => throw new IllegalStateException("Should only handle IP quota entities here")
     }
 
-    quotaDelta.changes().entrySet().forEach { e =>
+    quotaDelta.changes().forEach { (key, value) =>
       // The connection quota only understands the connection rate limit
-      val quotaName = e.getKey
-      val quotaValue = e.getValue
+      val quotaName = key
+      val quotaValue = value
       if (!quotaName.equals(QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG)) {
         warn(s"Ignoring unexpected quota key $quotaName for entity $ipEntity")
       } else {

diff --git a/core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala b/core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala
@@ -488,9 +488,9 @@ class KRaftMetadataCache(val brokerId: Int) extends MetadataCache with Logging w
     val internalTopics = new util.HashSet[String]
 
     image.topics().topicsByName().values().forEach { topic =>
-      topic.partitions().entrySet().forEach { entry =>
-        val partitionId = entry.getKey
-        val partition = entry.getValue
+      topic.partitions().forEach { (key, value) =>
+        val partitionId = key
+        val partition = value
         partitionInfos.add(new PartitionInfo(topic.name(),
           partitionId,
           node(partition.leader),

diff --git a/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala b/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala
@@ -346,8 +346,8 @@ abstract class KafkaServerTestHarness extends QuorumTestHarness {
   def getTopicNames(): Map[Uuid, String] = {
     if (isKRaftTest()) {
       val result = new util.HashMap[Uuid, String]()
-      controllerServer.controller.findAllTopicIds(ANONYMOUS_CONTEXT).get().entrySet().forEach {
-        e => result.put(e.getValue(), e.getKey())
+      controllerServer.controller.findAllTopicIds(ANONYMOUS_CONTEXT).get().forEach {
+        (key, value) => result.put(value, key)
       }
       result.asScala.toMap
     } else {

diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumFeatures.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumFeatures.java
@@ -141,7 +141,7 @@ public boolean equals(Object o) {
     @Override
     public String toString() {
         List<String> features = new ArrayList<>();
-        localSupportedFeatures.entrySet().forEach(f -> features.add(f.getKey() + ": " + f.getValue()));
+        localSupportedFeatures.forEach((key, value) -> features.add(key + ": " + value));
         features.sort(String::compareTo);
         List<String> nodeIds = new ArrayList<>();
         quorumNodeIds.forEach(id -> nodeIds.add("" + id));

diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
@@ -729,8 +729,7 @@ private ApiError createTopic(ControllerRequestContext context,
             }
             ApiError error = maybeCheckCreateTopicPolicy(() -> {
                 Map<Integer, List<Integer>> assignments = new HashMap<>();
-                newParts.entrySet().forEach(e -> assignments.put(e.getKey(),
-                    Replicas.toList(e.getValue().replicas)));
+                newParts.forEach((key, value) -> assignments.put(key, Replicas.toList(value.replicas)));
                 return new CreateTopicPolicy.RequestMetadata(
                     topic.name(), null, null, assignments, creationConfigs);
             });

diff --git a/metadata/src/main/java/org/apache/kafka/metadata/properties/MetaPropertiesEnsemble.java b/metadata/src/main/java/org/apache/kafka/metadata/properties/MetaPropertiesEnsemble.java
@@ -607,8 +607,7 @@ public String toString() {
         TreeMap<String, String> outputMap = new TreeMap<>();
         emptyLogDirs.forEach(e -> outputMap.put(e, "EMPTY"));
         errorLogDirs.forEach(e -> outputMap.put(e, "ERROR"));
-        logDirProps.entrySet().forEach(
-            e -> outputMap.put(e.getKey(), e.getValue().toString()));
+        logDirProps.forEach((key, value) -> outputMap.put(key, value.toString()));
         return "MetaPropertiesEnsemble" +
             "(metadataLogDir=" + metadataLogDir +
             ", dirs={" + outputMap.entrySet().stream().

diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
@@ -304,9 +304,12 @@ CreatableTopicResult createTestTopic(String name, int[][] replicas,
                 topic.assignments().add(new CreatableReplicaAssignment().
                     setPartitionIndex(i).setBrokerIds(Replicas.toList(replicas[i])));
             }
-            configs.entrySet().forEach(e -> topic.configs().add(
-                new CreateTopicsRequestData.CreateableTopicConfig().setName(e.getKey()).
-                    setValue(e.getValue())));
+            configs.forEach((key, value) -> topic.configs().add(
+                    new CreateTopicsRequestData.CreateableTopicConfig()
+                            .setName(key)
+                            .setValue(value)
+                    )
+            );
             request.topics().add(topic);
             ControllerRequestContext requestContext = anonymousContextFor(ApiKeys.CREATE_TOPICS);
             ControllerResult<CreateTopicsResponseData> result =

diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java
@@ -314,10 +314,9 @@ synchronized void onChange(final Map<HostInfo, Set<TopicPartition>> activePartit
                                final Map<HostInfo, Set<TopicPartition>> standbyPartitionHostMap,
                                final Map<TopicPartition, PartitionInfo> topicPartitionInfo) {
         this.partitionsByTopic = new HashMap<>();
-        topicPartitionInfo.entrySet().forEach(entry -> this.partitionsByTopic
-                .computeIfAbsent(entry.getKey().topic(), topic -> new ArrayList<>())
-                .add(entry.getValue())
-        );
+        topicPartitionInfo.forEach((key, value) -> this.partitionsByTopic
+                .computeIfAbsent(key.topic(), topic -> new ArrayList<>())
+                .add(value));
 
         rebuildMetadata(activePartitionHostMap, standbyPartitionHostMap);
     }

diff --git a/tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java b/tools/src/main/java/org/apache/kafka/tools/LeaderElectionCommand.java
@@ -86,7 +86,7 @@ static void run(Duration timeoutMs, String... args) throws Exception {
         ElectionType electionType = commandOptions.getElectionType();
         Optional<Set<TopicPartition>> jsonFileTopicPartitions =
             Optional.ofNullable(commandOptions.getPathToJsonFile())
-                .map(path -> parseReplicaElectionData(path));
+                .map(LeaderElectionCommand::parseReplicaElectionData);
 
         Optional<String> topicOption = Optional.ofNullable(commandOptions.getTopic());
         Optional<Integer> partitionOption = Optional.ofNullable(commandOptions.getPartition());
@@ -144,16 +144,15 @@ private static void electLeaders(Admin client, ElectionType electionType, Option
         Set<TopicPartition> noop = new HashSet<>();
         Map<TopicPartition, Throwable> failed = new HashMap<>();
 
-        electionResults.entrySet().stream().forEach(entry -> {
-            Optional<Throwable> error = entry.getValue();
+        electionResults.forEach((key, error) -> {
             if (error.isPresent()) {
                 if (error.get() instanceof ElectionNotNeededException) {
-                    noop.add(entry.getKey());
+                    noop.add(key);
                 } else {
-                    failed.put(entry.getKey(), error.get());
+                    failed.put(key, error.get());
                 }
             } else {
-                succeeded.add(entry.getKey());
+                succeeded.add(key);
             }
         });
 
@@ -175,10 +174,16 @@ private static void electLeaders(Admin client, ElectionType electionType, Option
         if (!failed.isEmpty()) {
             AdminCommandFailedException rootException =
                 new AdminCommandFailedException(String.format("%s replica(s) could not be elected", failed.size()));
-            failed.entrySet().forEach(entry -> {
-                System.out.println(String.format("Error completing leader election (%s) for partition: %s: %s",
-                    electionType, entry.getKey(), entry.getValue()));
-                rootException.addSuppressed(entry.getValue());
+            failed.forEach((key, value) -> {
+                System.out.println(
+                        String.format(
+                                "Error completing leader election (%s) for partition: %s: %s",
+                                electionType,
+                                key,
+                                value
+                        )
+                );
+                rootException.addSuppressed(value);
             });
             throw rootException;
         }