KAFKA-19139 Plugin#wrapInstance should use LinkedHashMap instead of Map (#19519)

There will be an update to the PluginMetrics#metricName method: the type
of the tags parameter will be changed
from Map to LinkedHashMap.
This change is necessary because the order of metric tags is important
1. If the tag order is inconsistent, identical metrics may be treated as
distinct ones by the metrics backend
2. KAFKA-18390 is updating metric naming to use LinkedHashMap. For
consistency, we should follow the same approach here.

Reviewers: TengYao Chi <[email protected]>, Jhen-Yung Hsu
 <[email protected]>, lllilllilllilili

Author: m1a2st

clients/clients-integration-tests/src/test/java/org/apache/kafka/server/quota/CustomQuotaCallbackTest.java

@@ -173,7 +173,7 @@ public static class MonitorableCustomQuotaCallback extends CustomQuotaCallback i
 
         @Override
         public void withPluginMetrics(PluginMetrics metrics) {
-            MetricName metricName = metrics.metricName(METRIC_NAME, METRIC_DESCRIPTION, Map.of());
+            MetricName metricName = metrics.metricName(METRIC_NAME, METRIC_DESCRIPTION, new LinkedHashMap<>());
             metrics.addMetric(metricName, (Gauge<Integer>) (config, now) -> 1);
         }
 

clients/src/main/java/org/apache/kafka/common/metrics/PluginMetrics.java

@@ -18,7 +18,7 @@
 
 import org.apache.kafka.common.MetricName;
 
-import java.util.Map;
+import java.util.LinkedHashMap;
 
 /**
  * This allows plugins to register metrics and sensors.
@@ -35,7 +35,7 @@ public interface PluginMetrics {
      * @param tags        Additional tags for the metric
      * @throws IllegalArgumentException if any of the tag names collide with the default tags for the plugin
      */
-    MetricName metricName(String name, String description, Map<String, String> tags);
+    MetricName metricName(String name, String description, LinkedHashMap<String, String> tags);
 
     /**
      * Add a metric to monitor an object that implements {@link MetricValueProvider}. This metric won't be associated with any

clients/src/main/java/org/apache/kafka/common/metrics/internals/PluginMetricsImpl.java

@@ -45,7 +45,7 @@ public PluginMetricsImpl(Metrics metrics, Map<String, String> tags) {
     }
 
     @Override
-    public MetricName metricName(String name, String description, Map<String, String> tags) {
+    public MetricName metricName(String name, String description, LinkedHashMap<String, String> tags) {
         if (closing) throw new IllegalStateException("This PluginMetrics instance is closed");
         for (String tagName : tags.keySet()) {
             if (this.tags.containsKey(tagName)) {

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java

@@ -3746,9 +3746,13 @@ private MetricName expectedMetricName(String clientId, String config, Class<?> c
 
     private static final String NAME = "name";
     private static final String DESCRIPTION = "description";
-    private static final Map<String, String> TAGS = Collections.singletonMap("k", "v");
+    private static final LinkedHashMap<String, String> TAGS = new LinkedHashMap<>();
     private static final double VALUE = 123.0;
 
+    static {
+        TAGS.put("t1", "v1");
+    }
+
     public static class MonitorableDeserializer extends MockDeserializer implements Monitorable {
 
         @Override

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

@@ -238,14 +238,7 @@ public void testOverwriteAcksAndRetriesForIdempotentProducers() {
 
     @Test
     public void testAcksAndIdempotenceForIdempotentProducers() {
-        Properties baseProps = new Properties() {{
-                setProperty(
-                    ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
-                setProperty(
-                    ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
-                setProperty(
-                    ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
-            }};
+        Properties baseProps = baseProperties();
 
         Properties validProps = new Properties() {{
                 putAll(baseProps);
@@ -348,11 +341,7 @@ public void testAcksAndIdempotenceForIdempotentProducers() {
 
     @Test
     public void testRetriesAndIdempotenceForIdempotentProducers() {
-        Properties baseProps = new Properties() {{
-                setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
-                setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
-                setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
-            }};
+        Properties baseProps = baseProperties();
 
         Properties validProps = new Properties() {{
                 putAll(baseProps);
@@ -414,13 +403,17 @@ public void testRetriesAndIdempotenceForIdempotentProducers() {
             "Must set retries to non-zero when using the transactional producer.");
     }
 
+    private Properties baseProperties() {
+        Properties baseProps = new Properties();
+        baseProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+        baseProps.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        baseProps.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        return baseProps;
+    }
+
     @Test
     public void testInflightRequestsAndIdempotenceForIdempotentProducers() {
-        Properties baseProps = new Properties() {{
-                setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
-                setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
-                setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
-            }};
+        Properties baseProps = baseProperties();
 
         Properties validProps = new Properties() {{
                 putAll(baseProps);
@@ -1590,7 +1583,7 @@ public void testMeasureAbortTransactionDuration() {
     }
 
     @Test
-    public void testCommitTransactionWithRecordTooLargeException() throws Exception {
+    public void testCommitTransactionWithRecordTooLargeException() {
         Map<String, Object> configs = new HashMap<>();
         configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
         configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
@@ -1620,7 +1613,7 @@ public void testCommitTransactionWithRecordTooLargeException() throws Exception
     }
 
     @Test
-    public void testCommitTransactionWithMetadataTimeoutForMissingTopic() throws Exception {
+    public void testCommitTransactionWithMetadataTimeoutForMissingTopic() {
         Map<String, Object> configs = new HashMap<>();
         configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
         configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
@@ -1657,7 +1650,7 @@ public void testCommitTransactionWithMetadataTimeoutForMissingTopic() throws Exc
     }
 
     @Test
-    public void testCommitTransactionWithMetadataTimeoutForPartitionOutOfRange() throws Exception {
+    public void testCommitTransactionWithMetadataTimeoutForPartitionOutOfRange() {
         Map<String, Object> configs = new HashMap<>();
         configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
         configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
@@ -1694,7 +1687,7 @@ public void testCommitTransactionWithMetadataTimeoutForPartitionOutOfRange() thr
     }
 
     @Test
-    public void testCommitTransactionWithSendToInvalidTopic() throws Exception {
+    public void testCommitTransactionWithSendToInvalidTopic() {
         Map<String, Object> configs = new HashMap<>();
         configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "some.id");
         configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
@@ -2131,7 +2124,7 @@ public void testOnlyCanExecuteCloseAfterInitTransactionsTimeout() {
     }
 
     @Test
-    public void testSendToInvalidTopic() throws Exception {
+    public void testSendToInvalidTopic() {
         Map<String, Object> configs = new HashMap<>();
         configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
         configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "15000");
@@ -2937,9 +2930,13 @@ private MetricName expectedMetricName(String clientId, String config, Class<?> c
 
     private static final String NAME = "name";
     private static final String DESCRIPTION = "description";
-    private static final Map<String, String> TAGS = Collections.singletonMap("k", "v");
+    private static final LinkedHashMap<String, String> TAGS = new LinkedHashMap<>();
     private static final double VALUE = 123.0;
 
+    static {
+        TAGS.put("t1", "v1");
+    }
+
     public static class MonitorableSerializer extends MockSerializer implements Monitorable {
 
         @Override

clients/src/test/java/org/apache/kafka/common/internals/PluginTest.java

@@ -26,7 +26,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Arrays;
-import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -129,7 +129,7 @@ public void testUsePluginMetricsAfterClose() throws Exception {
         Plugin<SomeMonitorablePlugin> plugin = Plugin.wrapInstance(new SomeMonitorablePlugin(), METRICS, CONFIG);
         PluginMetrics pluginMetrics = plugin.get().pluginMetrics;
         plugin.close();
-        assertThrows(IllegalStateException.class, () -> pluginMetrics.metricName("", "", Collections.emptyMap()));
+        assertThrows(IllegalStateException.class, () -> pluginMetrics.metricName("", "", new LinkedHashMap<>()));
         assertThrows(IllegalStateException.class, () -> pluginMetrics.addMetric(null, null));
         assertThrows(IllegalStateException.class, () -> pluginMetrics.removeMetric(null));
         assertThrows(IllegalStateException.class, () -> pluginMetrics.addSensor(""));

clients/src/test/java/org/apache/kafka/common/metrics/internals/PluginMetricsImplTest.java

@@ -27,7 +27,6 @@
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
-import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
@@ -36,11 +35,15 @@
 
 public class PluginMetricsImplTest {
 
-    private final Map<String, String> extraTags = Collections.singletonMap("my-tag", "my-value");
+    private static final LinkedHashMap<String, String> EXTRA_TAGS = new LinkedHashMap<>();
     private Map<String, String> tags;
     private Metrics metrics;
     private int initialMetrics;
 
+    static {
+        EXTRA_TAGS.put("my-tag", "my-value");
+    }
+
     @BeforeEach
     void setup() {
         metrics = new Metrics();
@@ -53,26 +56,28 @@ void setup() {
     @Test
     void testMetricName() {
         PluginMetricsImpl pmi = new PluginMetricsImpl(metrics, tags);
-        MetricName metricName = pmi.metricName("name", "description", extraTags);
+        MetricName metricName = pmi.metricName("name", "description", EXTRA_TAGS);
         assertEquals("name", metricName.name());
         assertEquals("plugins", metricName.group());
         assertEquals("description", metricName.description());
         Map<String, String> expectedTags = new LinkedHashMap<>(tags);
-        expectedTags.putAll(extraTags);
+        expectedTags.putAll(EXTRA_TAGS);
         assertEquals(expectedTags, metricName.tags());
     }
 
     @Test
     void testDuplicateTagName() {
         PluginMetricsImpl pmi = new PluginMetricsImpl(metrics, tags);
+        LinkedHashMap<String, String> tags = new LinkedHashMap<>();
+        tags.put("k1", "value");
         assertThrows(IllegalArgumentException.class,
-                () -> pmi.metricName("name", "description", Collections.singletonMap("k1", "value")));
+                () -> pmi.metricName("name", "description", tags));
     }
 
     @Test
     void testAddRemoveMetrics() {
         PluginMetricsImpl pmi = new PluginMetricsImpl(metrics, tags);
-        MetricName metricName = pmi.metricName("name", "description", extraTags);
+        MetricName metricName = pmi.metricName("name", "description", EXTRA_TAGS);
         pmi.addMetric(metricName, (Measurable) (config, now) -> 0.0);
         assertEquals(initialMetrics + 1, metrics.metrics().size());
 
@@ -88,7 +93,7 @@ void testAddRemoveMetrics() {
     void testAddRemoveSensor() {
         PluginMetricsImpl pmi = new PluginMetricsImpl(metrics, tags);
         String sensorName = "my-sensor";
-        MetricName metricName = pmi.metricName("name", "description", extraTags);
+        MetricName metricName = pmi.metricName("name", "description", EXTRA_TAGS);
         Sensor sensor = pmi.addSensor(sensorName);
         assertEquals(initialMetrics, metrics.metrics().size());
         sensor.add(metricName, new Rate());
@@ -107,10 +112,10 @@ void testAddRemoveSensor() {
     void testClose() throws IOException {
         PluginMetricsImpl pmi = new PluginMetricsImpl(metrics, tags);
         String sensorName = "my-sensor";
-        MetricName metricName1 = pmi.metricName("name1", "description", extraTags);
+        MetricName metricName1 = pmi.metricName("name1", "description", EXTRA_TAGS);
         Sensor sensor = pmi.addSensor(sensorName);
         sensor.add(metricName1, new Rate());
-        MetricName metricName2 = pmi.metricName("name2", "description", extraTags);
+        MetricName metricName2 = pmi.metricName("name2", "description", EXTRA_TAGS);
         pmi.addMetric(metricName2, (Measurable) (config, now) -> 1.0);
 
         assertEquals(initialMetrics + 2, metrics.metrics().size());

connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkConnector.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.connect.sink.SinkRecord;
 
 import java.util.Collection;
+import java.util.LinkedHashMap;
 import java.util.Map;
 
 public class MonitorableSinkConnector extends TestableSinkConnector {
@@ -35,7 +36,7 @@ public class MonitorableSinkConnector extends TestableSinkConnector {
     public void start(Map<String, String> props) {
         super.start(props);
         PluginMetrics pluginMetrics = context.pluginMetrics();
-        metricsName = pluginMetrics.metricName("start", "description", Map.of());
+        metricsName = pluginMetrics.metricName("start", "description", new LinkedHashMap<>());
         pluginMetrics.addMetric(metricsName, (Gauge<Object>) (config, now) -> VALUE);
     }
 
@@ -53,7 +54,7 @@ public static class MonitorableSinkTask extends TestableSinkTask {
         public void start(Map<String, String> props) {
             super.start(props);
             PluginMetrics pluginMetrics = context.pluginMetrics();
-            metricsName = pluginMetrics.metricName("put", "description", Map.of());
+            metricsName = pluginMetrics.metricName("put", "description", new LinkedHashMap<>());
             pluginMetrics.addMetric(metricsName, (Measurable) (config, now) -> count);
         }
 

connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java

@@ -23,6 +23,7 @@
 import org.apache.kafka.connect.connector.Task;
 import org.apache.kafka.connect.source.SourceRecord;
 
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -35,7 +36,7 @@ public class MonitorableSourceConnector extends TestableSourceConnector {
     public void start(Map<String, String> props) {
         super.start(props);
         PluginMetrics pluginMetrics = context.pluginMetrics();
-        metricsName = pluginMetrics.metricName("start", "description", Map.of());
+        metricsName = pluginMetrics.metricName("start", "description", new LinkedHashMap<>());
         pluginMetrics.addMetric(metricsName, (Gauge<Object>) (config, now) -> VALUE);
     }
 
@@ -53,7 +54,7 @@ public static class MonitorableSourceTask extends TestableSourceTask {
         public void start(Map<String, String> props) {
             super.start(props);
             PluginMetrics pluginMetrics = context.pluginMetrics();
-            metricsName = pluginMetrics.metricName("poll", "description", Map.of());
+            metricsName = pluginMetrics.metricName("poll", "description", new LinkedHashMap<>());
             pluginMetrics.addMetric(metricsName, (Measurable) (config, now) -> count);
         }
 

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java

@@ -78,6 +78,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -799,7 +800,7 @@ protected boolean isAllowed(ConfigValue configValue) {
 
         @Override
         public void withPluginMetrics(PluginMetrics metrics) {
-            metricName = metrics.metricName("name", "description", Map.of());
+            metricName = metrics.metricName("name", "description", new LinkedHashMap<>());
             metrics.addMetric(metricName, (Measurable) (config, now) -> count);
         }
     }

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ConnectMetricsTest.java

@@ -64,10 +64,13 @@ public class ConnectMetricsTest {
         WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter",
         WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
     private static final ConnectorTaskId CONNECTOR_TASK_ID = new ConnectorTaskId("connector", 0);
-    private static final Map<String, String> TAGS = Map.of("t1", "v1");
-
+    private static final LinkedHashMap<String, String> TAGS = new LinkedHashMap<>();
     private ConnectMetrics metrics;
-
+    
+    static {
+        TAGS.put("t1", "v1");
+    }
+    
     @BeforeEach
     public void setUp() {
         metrics = new ConnectMetrics("worker1", new WorkerConfig(WorkerConfig.baseConfigDef(), DEFAULT_WORKER_CONFIG), new MockTime(), "cluster-1");

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/ConnectRestServerTest.java

@@ -64,6 +64,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -402,7 +403,7 @@ public void register(ConnectRestExtensionContext restPluginContext) {
 
         @Override
         public void withPluginMetrics(PluginMetrics metrics) {
-            metricName = metrics.metricName("name", "description", Map.of());
+            metricName = metrics.metricName("name", "description", new LinkedHashMap<>());
             metrics.addMetric(metricName, (Gauge<Boolean>) (config, now) -> called);
         }
     }

metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java

@@ -39,6 +39,7 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -230,21 +231,21 @@ private class AuthorizerMetrics {
         private AuthorizerMetrics(PluginMetrics metrics) {
             authorizationAllowedSensor = metrics.addSensor("authorizer-authorization-allowed");
             authorizationAllowedSensor.add(
-                    metrics.metricName("authorization-allowed-rate-per-minute", "The number of authorization allowed per minute", Map.of()),
+                    metrics.metricName("authorization-allowed-rate-per-minute", "The number of authorization allowed per minute", new LinkedHashMap<>()),
                     new Rate(TimeUnit.MINUTES, new WindowedCount()));
 
             authorizationDeniedSensor = metrics.addSensor("authorizer-authorization-denied");
             authorizationDeniedSensor.add(
-                    metrics.metricName("authorization-denied-rate-per-minute", "The number of authorization denied per minute", Map.of()),
+                    metrics.metricName("authorization-denied-rate-per-minute", "The number of authorization denied per minute", new LinkedHashMap<>()),
                     new Rate(TimeUnit.MINUTES, new WindowedCount()));
 
             authorizationRequestSensor = metrics.addSensor("authorizer-authorization-request");
             authorizationRequestSensor.add(
-                    metrics.metricName("authorization-request-rate-per-minute", "The number of authorization request per minute", Map.of()),
+                    metrics.metricName("authorization-request-rate-per-minute", "The number of authorization request per minute", new LinkedHashMap<>()),
                     new Rate(TimeUnit.MINUTES, new WindowedCount()));
 
             metrics.addMetric(
-                    metrics.metricName("acls-total-count", "The number of acls defined", Map.of()),
+                    metrics.metricName("acls-total-count", "The number of acls defined", new LinkedHashMap<>()),
                     (Gauge<Integer>) (config, now) -> aclCount());
         }