KAFKA-17016: Align the behavior of GaugeWrapper and MeterWrapper

[FrankYang0529]

When using MeterWrapper, it initializes Meter again after users close it.
However, the GaugeWrapper behavior is different. We use aggregatedMetric directly.
If GaugeWrapper is closed, we don't have chance to initialize it again.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[chia7712]

@FrankYang0529 thanks for this patch!

[chia7712]

@FrankYang0529 thanks for your patch

[chia7712]

it seems the lock is used to make sure we don't call removeMetric/newGauge repeatedly, so we can do a bit refactor for it. WDYT?

  case class GaugeWrapper(metricType: String) {
    private final val removed = new AtomicBoolean(false)
    // The map to store:
    //   - per-partition value for topic-level metrics. The key will be the partition number
    //   - per-topic value for broker-level metrics. The key will be the topic name
    private val metricValues = new ConcurrentHashMap[String, Long]()

    def setValue(key: String, value: Long): Unit = {
      newGaugeIfNeed()
      metricValues.put(key, value)
    }

    def removeKey(key: String): Unit = {
      newGaugeIfNeed()
      metricValues.remove(key)
    }

    def close(): Unit = if (removed.compareAndSet(true, false)) {
      metricsGroup.removeMetric(metricType, tags)
      metricValues.clear()
    }

    private def newGaugeIfNeed(): Unit = if (removed.compareAndSet(false, true)) {
      metricsGroup.newGauge(metricType, () => metricValues.values().stream().mapToLong(v => v).sum(), tags)
    }

    newGaugeIfNeed()
  }

[FrankYang0529]

Thanks for the great suggestion. Updated it.