MINOR: use enum map for error counts map (#19314)

Java provides a specialised Map where Enums are the keys, which can
provide some performance improvements.

https://docs.oracle.com/javase/8/docs/api/java/util/EnumMap.html

I have updated the Java code where possible to use an EnumMap rather
than a HashMap and run the unit tests under the requests directory.

Reviewers: Matthias J. Sax <[email protected]>, Lianet Magrans
<[email protected]>, Chia-Ping Tsai <[email protected]>

Author: lorcanj

clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java

@@ -25,7 +25,7 @@
 import java.nio.ByteBuffer;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -71,14 +71,14 @@ protected static Map<Errors, Integer> errorCounts(Stream<Errors> errors) {
     }
 
     protected static Map<Errors, Integer> errorCounts(Collection<Errors> errors) {
-        Map<Errors, Integer> errorCounts = new HashMap<>();
+        Map<Errors, Integer> errorCounts = new EnumMap<>(Errors.class);
         for (Errors error : errors)
             updateErrorCounts(errorCounts, error);
         return errorCounts;
     }
 
     protected static Map<Errors, Integer> apiErrorCounts(Map<?, ApiError> errors) {
-        Map<Errors, Integer> errorCounts = new HashMap<>();
+        Map<Errors, Integer> errorCounts = new EnumMap<>(Errors.class);
         for (ApiError apiError : errors.values())
             updateErrorCounts(errorCounts, apiError.error());
         return errorCounts;

clients/src/main/java/org/apache/kafka/common/requests/AlterClientQuotasResponse.java

@@ -27,6 +27,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -74,7 +75,7 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> counts = new HashMap<>();
+        Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         data.entries().forEach(entry ->
             updateErrorCounts(counts, Errors.forCode(entry.errorCode()))
         );

clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionReassignmentsResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 public class AlterPartitionReassignmentsResponse extends AbstractResponse {
@@ -62,7 +62,7 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> counts = new HashMap<>();
+        Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         updateErrorCounts(counts, Errors.forCode(data.errorCode()));
 
         data.responses().forEach(topicResponse ->

clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 public class AlterPartitionResponse extends AbstractResponse {
@@ -42,7 +42,7 @@ public AlterPartitionResponseData data() {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> counts = new HashMap<>();
+        Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         updateErrorCounts(counts, Errors.forCode(data.errorCode()));
         data.topics().forEach(topicResponse -> topicResponse.partitions().forEach(partitionResponse ->
             updateErrorCounts(counts, Errors.forCode(partitionResponse.errorCode()))

clients/src/main/java/org/apache/kafka/common/requests/AlterReplicaLogDirsResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 /**
@@ -60,7 +60,7 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> errorCounts = new HashMap<>();
+        Map<Errors, Integer> errorCounts = new EnumMap<>(Errors.class);
         data.results().forEach(topicResult ->
             topicResult.partitions().forEach(partitionResult ->
                 updateErrorCounts(errorCounts, Errors.forCode(partitionResult.errorCode()))));

clients/src/main/java/org/apache/kafka/common/requests/AlterShareGroupOffsetsResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 public class AlterShareGroupOffsetsResponse extends AbstractResponse {
@@ -37,7 +37,7 @@ public AlterShareGroupOffsetsResponse(AlterShareGroupOffsetsResponseData data) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> counts = new HashMap<>();
+        Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         data.responses().forEach(topic -> topic.partitions().forEach(partitionResponse ->
             updateErrorCounts(counts, Errors.forCode(partitionResponse.errorCode()))
         ));

clients/src/main/java/org/apache/kafka/common/requests/BeginQuorumEpochResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 /**
@@ -49,7 +49,7 @@ public BeginQuorumEpochResponse(BeginQuorumEpochResponseData data) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> errors = new HashMap<>();
+        Map<Errors, Integer> errors = new EnumMap<>(Errors.class);
 
         errors.put(Errors.forCode(data.errorCode()), 1);
 

clients/src/main/java/org/apache/kafka/common/requests/BrokerHeartbeatResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 public class BrokerHeartbeatResponse extends AbstractResponse {
@@ -51,7 +51,7 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> errorCounts = new HashMap<>();
+        Map<Errors, Integer> errorCounts = new EnumMap<>(Errors.class);
         errorCounts.put(Errors.forCode(data.errorCode()), 1);
         return errorCounts;
     }

clients/src/main/java/org/apache/kafka/common/requests/BrokerRegistrationResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 public class BrokerRegistrationResponse extends AbstractResponse {
@@ -51,7 +51,7 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> errorCounts = new HashMap<>();
+        Map<Errors, Integer> errorCounts = new EnumMap<>(Errors.class);
         errorCounts.put(Errors.forCode(data.errorCode()), 1);
         return errorCounts;
     }

clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupDescribeResponse.java

@@ -22,7 +22,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 /**
@@ -53,7 +53,7 @@ public ConsumerGroupDescribeResponseData data() {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        HashMap<Errors, Integer> counts = new HashMap<>();
+        Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         data.groups().forEach(
             group -> updateErrorCounts(counts, Errors.forCode(group.errorCode()))
         );

clients/src/main/java/org/apache/kafka/common/requests/CreatePartitionsResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 public class CreatePartitionsResponse extends AbstractResponse {
@@ -42,7 +42,7 @@ public CreatePartitionsResponseData data() {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> counts = new HashMap<>();
+        Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         data.results().forEach(result ->
             updateErrorCounts(counts, Errors.forCode(result.errorCode()))
         );

clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 public class CreateTopicsResponse extends AbstractResponse {
@@ -67,7 +67,7 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        HashMap<Errors, Integer> counts = new HashMap<>();
+        Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         data.topics().forEach(result ->
             updateErrorCounts(counts, Errors.forCode(result.errorCode()))
         );

clients/src/main/java/org/apache/kafka/common/requests/DeleteGroupsResponse.java

@@ -23,6 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -69,7 +70,7 @@ public Errors get(String group) throws IllegalArgumentException {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> counts = new HashMap<>();
+        Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         data.results().forEach(result ->
             updateErrorCounts(counts, Errors.forCode(result.errorCode()))
         );

clients/src/main/java/org/apache/kafka/common/requests/DeleteRecordsResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 public class DeleteRecordsResponse extends AbstractResponse {
@@ -63,7 +63,7 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> errorCounts = new HashMap<>();
+        Map<Errors, Integer> errorCounts = new EnumMap<>(Errors.class);
         data.topics().forEach(topicResponses ->
             topicResponses.partitions().forEach(response ->
                 updateErrorCounts(errorCounts, Errors.forCode(response.errorCode()))

clients/src/main/java/org/apache/kafka/common/requests/DeleteShareGroupOffsetsResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 public class DeleteShareGroupOffsetsResponse extends AbstractResponse {
@@ -41,7 +41,7 @@ public DeleteShareGroupOffsetsResponseData data() {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> counts = new HashMap<>();
+        Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         data.responses().forEach(
             topicResult -> topicResult.partitions().forEach(
                 partitionResult -> updateErrorCounts(counts, Errors.forCode(partitionResult.errorCode()))

clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java

@@ -22,7 +22,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 
@@ -62,7 +62,7 @@ public DeleteTopicsResponseData data() {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        HashMap<Errors, Integer> counts = new HashMap<>();
+        Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         data.responses().forEach(result ->
             updateErrorCounts(counts, Errors.forCode(result.errorCode()))
         );

clients/src/main/java/org/apache/kafka/common/requests/DescribeConfigsResponse.java

@@ -25,7 +25,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.Collection;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.Function;
@@ -241,7 +241,7 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> errorCounts = new HashMap<>();
+        Map<Errors, Integer> errorCounts = new EnumMap<>(Errors.class);
         data.results().forEach(response ->
             updateErrorCounts(errorCounts, Errors.forCode(response.errorCode()))
         );

clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsResponse.java

@@ -26,7 +26,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.Collections;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -139,7 +139,7 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> errorCounts = new HashMap<>();
+        Map<Errors, Integer> errorCounts = new EnumMap<>(Errors.class);
         data.groups().forEach(describedGroup ->
             updateErrorCounts(errorCounts, Errors.forCode(describedGroup.errorCode())));
         return errorCounts;

clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java

@@ -23,7 +23,7 @@
 import org.apache.kafka.common.protocol.Errors;
 
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 
@@ -56,7 +56,7 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> errorCounts = new HashMap<>();
+        Map<Errors, Integer> errorCounts = new EnumMap<>(Errors.class);
         errorCounts.put(Errors.forCode(data.errorCode()), 1);
         data.results().forEach(result ->
             updateErrorCounts(errorCounts, Errors.forCode(result.errorCode()))

clients/src/main/java/org/apache/kafka/common/requests/DescribeQuorumResponse.java

@@ -24,7 +24,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.Collections;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.Map;
 
 /**
@@ -48,7 +48,7 @@ public DescribeQuorumResponse(DescribeQuorumResponseData data) {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> errors = new HashMap<>();
+        Map<Errors, Integer> errors = new EnumMap<>(Errors.class);
 
         errors.put(Errors.forCode(data.errorCode()), 1);
 

clients/src/main/java/org/apache/kafka/common/requests/DescribeShareGroupOffsetsResponse.java

@@ -25,6 +25,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -77,7 +78,7 @@ public DescribeShareGroupOffsetsResponseData data() {
 
     @Override
     public Map<Errors, Integer> errorCounts() {
-        Map<Errors, Integer> counts = new HashMap<>();
+        Map<Errors, Integer> counts = new EnumMap<>(Errors.class);
         groupLevelErrors.values().forEach(exception -> updateErrorCounts(counts, Errors.forException(exception)));
         for (DescribeShareGroupOffsetsResponseGroup group : data.groups()) {
             group.topics().forEach(topic ->