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 398819016cb5b..a7581442f5d12 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
@@ -502,6 +502,7 @@ public void addMetric(MetricName metricName, MetricConfig config, Measurable mea
      *
      * @param metricName The name of the metric
      * @param metricValueProvider The metric value provider associated with this metric
+     * @throws IllegalArgumentException if a metric with same name already exists.
      */
     public void addMetric(MetricName metricName, MetricConfig config, MetricValueProvider<?> metricValueProvider) {
         KafkaMetric m = new KafkaMetric(new Object(),
@@ -587,18 +588,19 @@ public synchronized void removeReporter(MetricsReporter reporter) {
     }
 
     /**
-     * Register a metric if not present or return an already existing metric otherwise.
+     * Register a metric if not present or return the already existing metric with the same name.
      * When a metric is newly registered, this method returns null
      *
      * @param metric The KafkaMetric to register
-     * @return KafkaMetric if the metric already exists, null otherwise
+     * @return the existing metric with the same name or null
      */
     synchronized KafkaMetric registerMetric(KafkaMetric metric) {
         MetricName metricName = metric.metricName();
-        if (this.metrics.containsKey(metricName)) {
-            return this.metrics.get(metricName);
+        KafkaMetric existingMetric = this.metrics.putIfAbsent(metricName, metric);
+        if (existingMetric != null) {
+            return existingMetric;
         }
-        this.metrics.put(metricName, metric);
+        // newly added metric
         for (MetricsReporter reporter : reporters) {
             try {
                 reporter.metricChange(metric);
diff --git a/docs/upgrade.html b/docs/upgrade.html
index b29259aa2b8b0..498944646c85e 100644
--- a/docs/upgrade.html
+++ b/docs/upgrade.html
@@ -19,6 +19,15 @@
 
 <script id="upgrade-template" type="text/x-handlebars-template">
 
+<h5><a id="upgrade_330_notable" href="#upgrade_330_notable">Notable changes in 3.3.0</a></h5>
+    <ul>
+        <li>Introduced a new API <code>addMetricIfAbsent</code> to <code>Metrics</code> which would create a new Metric if not existing or return the same metric
+            if already registered. Note that this behaviour is different from <code>addMetric</code> API which throws an <code>IllegalArgumentException</code> when
+            trying to create an already existing metric. (See <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-843%3A+Adding+addMetricIfAbsent+method+to+Metrics">KIP-843</a>
+            for more details).
+        </li>
+    </ul>
+
 <h4><a id="upgrade_3_2_0" href="#upgrade_3_2_0">Upgrading to 3.2.0 from any version 0.8.x through 3.1.x</a></h4>
 
 <p><b>If you are upgrading from a version prior to 2.1.x, please see the note below about the change to the schema used to store consumer offsets.
@@ -68,7 +77,7 @@ <h5><a id="upgrade_320_notable" href="#upgrade_320_notable">Notable changes in 3
             (See <a href="https://issues.apache.org/jira/browse/KAFKA-13598">KAFKA-13598</a>for more details).
             This issue was fixed and the default is properly applied in 3.0.1, 3.1.1, and 3.2.0.</li>
         <li>A notable exception is Connect that by default disables idempotent behavior for all of its
-            producers in order to uniformly support using a wide range of Kafka broker versions. 
+            producers in order to uniformly support using a wide range of Kafka broker versions.
             Users can change this behavior to enable idempotence for some or all producers
             via Connect worker and/or connector configuration. Connect may enable idempotent producers
             by default in a future major release.</li>
@@ -80,8 +89,8 @@ <h5><a id="upgrade_320_notable" href="#upgrade_320_notable">Notable changes in 3
             <a href="https://www.slf4j.org/manual.html#swapping">slf4j-log4j12 version 1.7.35 or above</a> or
             slf4j-reload4j to avoid
             <a href="https://www.slf4j.org/codes.html#no_tlm">possible compatibility issues originating from the logging framework</a>.</li>
-        <li>The example connectors, <code>FileStreamSourceConnector</code> and <code>FileStreamSinkConnector</code>, have been 
-            removed from the default classpath. To use them in Kafka Connect standalone or distributed mode they need to be 
+        <li>The example connectors, <code>FileStreamSourceConnector</code> and <code>FileStreamSinkConnector</code>, have been
+            removed from the default classpath. To use them in Kafka Connect standalone or distributed mode they need to be
             explicitly added, for example <code>CLASSPATH=./lib/connect-file-3.2.0.jar ./bin/connect-distributed.sh</code>.</li>
     </ul>