fix ConsumerCoordinatorTest unit test

Author: mqliang

pinot-common/src/main/java/org/apache/pinot/common/metrics/AbstractMetrics.java

@@ -22,6 +22,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -236,13 +237,8 @@ public void addTimedValue(final String key, final T timer, final long duration,
    */
   private void addValueToTimer(String fullTimerName, String simplifiedTimerName, final long duration,
       final TimeUnit timeUnit, Map<String, String> attributes) {
-    Map<String, String> fullAttributes = ImmutableMap.<String, String>builder()
-        .putAll(attributes)
-        .put(MetricAttributeConstants.PINOT_METRIC_NAME, fullTimerName)
-        .build();
-
     final PinotMetricName metricName = PinotMetricUtils.makePinotMetricName(_clazz, fullTimerName,
-        simplifiedTimerName, fullAttributes);
+        simplifiedTimerName, new HashMap<>(attributes));
     PinotTimer timer =
         PinotMetricUtils.makePinotTimer(_metricsRegistry, metricName, TimeUnit.MILLISECONDS, TimeUnit.SECONDS);
     if (timer != null) {
@@ -366,10 +362,9 @@ public PinotMeter addMeteredTableValue(final String tableName, final String key,
       PinotMeter reusedMeter, Map<String, String> attributes) {
     String meterName = meter.getMeterName();
     final String fullMeterName = _metricPrefix + getTableName(tableName) + "." + key + "." + meterName;
-    Map<String, String> fullAttributes = ImmutableMap.<String, String>builder()
-        .putAll(attributes)
-        .put(MetricAttributeConstants.TABLE_NAME, tableName)
-        .build();
+    Map<String, String> fullAttributes = new HashMap<>(attributes);
+    fullAttributes.put(MetricAttributeConstants.TABLE_NAME, tableName);
+
     return addValueToMeter(fullMeterName, meterName, meter.getUnit(), unitCount, reusedMeter, fullAttributes);
   }
 
@@ -392,13 +387,8 @@ private PinotMeter addValueToMeter(
       reusedMeter.mark(unitCount);
       return reusedMeter;
     } else {
-      Map<String, String> fullAttributes = ImmutableMap.<String, String>builder()
-          .putAll(attributes)
-          .put(MetricAttributeConstants.PINOT_METRIC_NAME, fullMeterName)
-          .build();
-
       final PinotMetricName metricName = PinotMetricUtils.makePinotMetricName(_clazz, fullMeterName,
-          simplifiedMeterName, fullAttributes);
+          simplifiedMeterName, new HashMap<>(attributes));
 
       final PinotMeter newMeter =
           PinotMetricUtils.makePinotMeter(_metricsRegistry, metricName, unit, TimeUnit.SECONDS);
@@ -688,12 +678,8 @@ public void addCallbackGauge(final String metricName,
       String simplifiedMetricName,
       Map<String, String> attributes,
       final Callable<Long> valueCallback) {
-    Map<String, String> fullAttributes = ImmutableMap.<String, String>builder()
-        .putAll(attributes)
-        .put(MetricAttributeConstants.PINOT_METRIC_NAME, metricName)
-        .build();
     PinotMetricName pinotMetricName = PinotMetricUtils.makePinotMetricName(_clazz, _metricPrefix + metricName,
-        simplifiedMetricName, fullAttributes);
+        simplifiedMetricName, new HashMap<>(attributes));
 
     PinotMetricUtils
         .makeGauge(_metricsRegistry, pinotMetricName,
@@ -737,10 +723,8 @@ public void setOrUpdateTableGauge(final String tableName, final String key, fina
       Map<String, String> attributes, final Supplier<Long> valueSupplier) {
     String fullGaugeName = composeTableGaugeName(tableName, key, gauge);
 
-    Map<String, String> fullAttributes = ImmutableMap.<String, String>builder()
-        .putAll(attributes)
-        .put(MetricAttributeConstants.TABLE_NAME, tableName)
-        .build();
+    Map<String, String> fullAttributes = new HashMap<>(attributes);
+    fullAttributes.put(MetricAttributeConstants.TABLE_NAME, tableName);
 
     setOrUpdateGauge(fullGaugeName, gauge.getGaugeName(), fullAttributes, valueSupplier::get);
   }
@@ -820,13 +804,8 @@ public void setOrUpdateGlobalGauge(final String metricName, final LongSupplier v
 
   public void setOrUpdateGauge(final String metricName, final String simplifiedMetricName,
       Map<String, String> attributes, final LongSupplier valueSupplier) {
-    Map<String, String> fullAttributes = ImmutableMap.<String, String>builder()
-        .putAll(attributes)
-        .put(MetricAttributeConstants.PINOT_METRIC_NAME, metricName)
-        .build();
-
     PinotMetricName pinotMetricName = PinotMetricUtils.makePinotMetricName(_clazz, _metricPrefix + metricName,
-        simplifiedMetricName, fullAttributes);
+        simplifiedMetricName, new HashMap<>(attributes));
 
     PinotGauge<Long> pinotGauge = PinotMetricUtils.makeGauge(_metricsRegistry, pinotMetricName,
         PinotMetricUtils.makePinotGauge(pinotMetricName, avoid -> valueSupplier.getAsLong()));

pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinatorTest.java

@@ -46,6 +46,7 @@ public FakeRealtimeTableDataManager(Semaphore segmentBuildSemaphore,
         boolean useIdealStateToCalculatePreviousSegment) {
       super(segmentBuildSemaphore);
       super._recentlyDeletedSegments = CacheBuilder.newBuilder().build();
+      super._tableNameWithType = "testTable";
       StreamIngestionConfig streamIngestionConfig = new StreamIngestionConfig(List.of(new HashMap<>()));
       streamIngestionConfig.setEnforceConsumptionInOrder(true);
       if (useIdealStateToCalculatePreviousSegment) {
@@ -147,7 +148,6 @@ public void testFirstConsumer()
     FakeConsumerCoordinator consumerCoordinator = new FakeConsumerCoordinator(true, realtimeTableDataManager);
     realtimeTableDataManager.setConsumerCoordinator(consumerCoordinator);
     ReentrantLock lock = (ReentrantLock) consumerCoordinator.getLock();
-    RealtimeSegmentDataManager mockedRealtimeSegmentDataManager = getMockedRealtimeSegmentDataManager();
     Map<String, String> serverSegmentStatusMap = new HashMap<>() {{
       put("server_1", "ONLINE");
       put("server_3", "ONLINE");

pinot-plugins/pinot-metrics/pinot-open-telemetry/src/main/java/org/apache/pinot/plugin/metrics/opentelemetry/OpenTelemetryMetricName.java

@@ -18,8 +18,9 @@
  */
 package org.apache.pinot.plugin.metrics.opentelemetry;
 
-import com.google.common.collect.ImmutableMap;
+import java.util.HashMap;
 import java.util.Map;
+import org.apache.pinot.common.metrics.MetricAttributeConstants;
 import org.apache.pinot.spi.metrics.PinotMetricName;
 
 
@@ -35,7 +36,8 @@ public class OpenTelemetryMetricName implements PinotMetricName {
   public OpenTelemetryMetricName(String fullMetricName, String simplifiedMetricName, Map<String, String> attributes) {
     _fullMetricName = fullMetricName;
     _simplifiedMetricName = simplifiedMetricName;
-    _attributes = ImmutableMap.copyOf(attributes);
+    _attributes = new HashMap<>(attributes);
+    _attributes.put(MetricAttributeConstants.PINOT_METRIC_NAME, fullMetricName);
   }
 
   public OpenTelemetryMetricName(PinotMetricName pinotMetricName) {

pinot-plugins/pinot-metrics/pinot-open-telemetry/src/main/java/org/apache/pinot/plugin/metrics/opentelemetry/OpenTelemetryUtil.java

@@ -52,7 +52,7 @@ public static Attributes toOpenTelemetryAttributes(Map<String, String> attribute
     AttributesBuilder attributesBuilder = Attributes.builder();
     if (attributes != null) {
       for (Map.Entry<String, String> entry : attributes.entrySet()) {
-        attributesBuilder.put(entry.getKey(), entry.getValue());
+        attributesBuilder.put(entry.getKey(), entry.getValue() == null ? "-" : entry.getValue());
       }
     }
     return attributesBuilder.build();