MINOR: Add DCL to improve performance (#22098)

Synchronized method will cause the process lock here every time, it
would be better to lock when we  actually need to create the meter,
otherwise we can just return the  existed value.

Reviewers: Ken Huang <s7133700@gmail.com>, Chia-Ping Tsai
 <chia7712@gmail.com>

Author: YutaLin

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

@@ -369,7 +369,7 @@ public Sensor sensor(String name, Sensor.RecordingLevel recordingLevel, Sensor..
      * @param parents The parent sensors
      * @return The sensor that is created
      */
-    public synchronized Sensor sensor(String name, MetricConfig config, Sensor... parents) {
+    public Sensor sensor(String name, MetricConfig config, Sensor... parents) {
         return this.sensor(name, config, Sensor.RecordingLevel.INFO, parents);
     }
 
@@ -383,7 +383,7 @@ public synchronized Sensor sensor(String name, MetricConfig config, Sensor... pa
      * @param parents The parent sensors
      * @return The sensor that is created
      */
-    public synchronized Sensor sensor(String name, MetricConfig config, Sensor.RecordingLevel recordingLevel, Sensor... parents) {
+    public Sensor sensor(String name, MetricConfig config, Sensor.RecordingLevel recordingLevel, Sensor... parents) {
         return sensor(name, config, Long.MAX_VALUE, recordingLevel, parents);
     }
 
@@ -398,18 +398,23 @@ public synchronized Sensor sensor(String name, MetricConfig config, Sensor.Recor
      * @param recordingLevel The recording level.
      * @return The sensor that is created
      */
-    public synchronized Sensor sensor(String name, MetricConfig config, long inactiveSensorExpirationTimeSeconds, Sensor.RecordingLevel recordingLevel, Sensor... parents) {
+    public Sensor sensor(String name, MetricConfig config, long inactiveSensorExpirationTimeSeconds, Sensor.RecordingLevel recordingLevel, Sensor... parents) {
         Sensor s = getSensor(name);
         if (s == null) {
-            s = new Sensor(this, name, parents, config == null ? this.config : config, time, inactiveSensorExpirationTimeSeconds, recordingLevel);
-            this.sensors.put(name, s);
-            if (parents != null) {
-                for (Sensor parent : parents) {
-                    List<Sensor> children = childrenSensors.computeIfAbsent(parent, k -> new ArrayList<>());
-                    children.add(s);
+            synchronized (this) {
+                s = getSensor(name);
+                if (s == null) {
+                    s = new Sensor(this, name, parents, config == null ? this.config : config, time, inactiveSensorExpirationTimeSeconds, recordingLevel);
+                    this.sensors.put(name, s);
+                    if (parents != null) {
+                        for (Sensor parent : parents) {
+                            List<Sensor> children = childrenSensors.computeIfAbsent(parent, k -> new ArrayList<>());
+                            children.add(s);
+                        }
+                    }
+                    log.trace("Added sensor with name {}", name);
                 }
             }
-            log.trace("Added sensor with name {}", name);
         }
         return s;
     }
@@ -424,7 +429,7 @@ public synchronized Sensor sensor(String name, MetricConfig config, long inactiv
      * @param parents The parent sensors
      * @return The sensor that is created
      */
-    public synchronized Sensor sensor(String name, MetricConfig config, long inactiveSensorExpirationTimeSeconds, Sensor... parents) {
+    public Sensor sensor(String name, MetricConfig config, long inactiveSensorExpirationTimeSeconds, Sensor... parents) {
         return this.sensor(name, config, inactiveSensorExpirationTimeSeconds, Sensor.RecordingLevel.INFO, parents);
     }
 

server-common/src/main/java/org/apache/kafka/server/util/timer/TimingWheel.java

@@ -128,15 +128,19 @@ public class TimingWheel {
         }
     }
 
-    private synchronized void addOverflowWheel() {
+    private void addOverflowWheel() {
         if (overflowWheel == null) {
-            overflowWheel = new TimingWheel(
-                interval,
-                wheelSize,
-                currentTimeMs,
-                taskCounter,
-                queue
-            );
+            synchronized (this) {
+                if (overflowWheel == null) {
+                    overflowWheel = new TimingWheel(
+                        interval,
+                        wheelSize,
+                        currentTimeMs,
+                        taskCounter,
+                        queue
+                    );
+                }
+            }
         }
     }