diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java b/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
index 23644784087c5..0f6dc885fb25a 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
@@ -41,6 +41,7 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
 
 import static java.util.Collections.emptyList;
 
@@ -192,9 +193,13 @@ public Metrics(MetricConfig defaultConfig, List<MetricsReporter> reporters, Time
      * @param tags        additional key/value attributes of the metric
      */
     public MetricName metricName(String name, String group, String description, Map<String, String> tags) {
-        Map<String, String> combinedTag = new LinkedHashMap<>(config.tags());
-        combinedTag.putAll(tags);
-        return new MetricName(name, group, description, combinedTag);
+        Map<String, String> combinedTag = new LinkedHashMap<>();
+        BiConsumer<String, String> combine = (k, v) -> {
+            combinedTag.put(k.intern(), v.intern());
+        };
+        config.tags().forEach(combine);
+        tags.forEach(combine);
+        return new MetricName(name.intern(), group.intern(), description, combinedTag);
     }
 
     /**
diff --git a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/TelemetryMetricNamingConvention.java b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/TelemetryMetricNamingConvention.java
index cb8c245a613a1..113b9a3dc64bb 100644
--- a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/TelemetryMetricNamingConvention.java
+++ b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/TelemetryMetricNamingConvention.java
@@ -19,12 +19,11 @@
 import org.apache.kafka.common.MetricName;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
-import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.regex.Pattern;
-import java.util.stream.Collectors;
 
 /**
  * This class encapsulates naming and mapping conventions defined as part of
@@ -37,6 +36,7 @@ public class TelemetryMetricNamingConvention {
 
     // remove metrics as it is redundant for telemetry metrics naming convention
     private static final Pattern GROUP_PATTERN = Pattern.compile("\\.(metrics)");
+    private static final Pattern DASH_PATTERN = Pattern.compile("-");
 
     public static MetricNamingStrategy<MetricName> getClientTelemetryMetricNamingStrategy(String prefix) {
         Objects.requireNonNull(prefix, "prefix cannot be null");
@@ -46,14 +46,17 @@ public static MetricNamingStrategy<MetricName> getClientTelemetryMetricNamingStr
             public MetricKey metricKey(MetricName metricName) {
                 Objects.requireNonNull(metricName, "metric name cannot be null");
 
-                return new MetricKey(fullMetricName(prefix, metricName.group(), metricName.name()),
+                return new MetricKey(
+                    fullMetricName(prefix, metricName.group(), metricName.name()).intern(),
                     Collections.unmodifiableMap(cleanTags(metricName.tags())));
             }
 
             @Override
             public MetricKey derivedMetricKey(MetricKey key, String derivedComponent) {
                 Objects.requireNonNull(derivedComponent, "derived component cannot be null");
-                return new MetricKey(key.name() + NAME_JOINER + derivedComponent, key.tags());
+                return new MetricKey(
+                    (key.name() + NAME_JOINER + derivedComponent).intern(),
+                    key.tags());
             }
         };
     }
@@ -111,14 +114,16 @@ private static String cleanMetric(String metric) {
      * @return the new map with keys replaced by snake_case representations.
      */
     private static Map<String, String> cleanTags(Map<String, String> raw) {
-        return raw.entrySet()
-            .stream()
-            .collect(Collectors.toMap(s -> clean(s.getKey(), TAG_JOINER), Entry::getValue));
+        Map<String, String> result = new HashMap<>();
+        raw.forEach((k, v) -> {
+            result.put(clean(k, TAG_JOINER).intern(), v.intern());
+        });
+        return result;
     }
 
     private static String clean(String raw, String joiner) {
         Objects.requireNonNull(raw, "metric data cannot be null");
         String lowerCase = raw.toLowerCase(Locale.ROOT);
-        return lowerCase.replaceAll("-", joiner);
+        return DASH_PATTERN.matcher(lowerCase).replaceAll(joiner);
     }
 }