Modify the OpenTelemetry ProtobufReader's Handling of Attribute Types

extensions-contrib/opentelemetry-extensions/src/main/java/org/apache/druid/data/input/opentelemetry/protobuf/OpenTelemetryMetricsProtobufReader.java

@@ -36,8 +36,10 @@
 import org.apache.druid.java.util.common.parsers.CloseableIterator;
 import org.apache.druid.java.util.common.parsers.ParseException;
 
+import javax.annotation.Nullable;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -94,8 +96,14 @@ private List<InputRow> parseMetricsData(final MetricsData metricsData)
           Map<String, Object> resourceAttributes = resourceMetrics.getResource()
               .getAttributesList()
               .stream()
-              .collect(Collectors.toMap(kv -> resourceAttributePrefix + kv.getKey(),
-                  kv -> parseAnyValue(kv.getValue())));
+              .collect(HashMap::new,
+                  (m, kv) -> {
+                    Object value = parseAnyValue(kv.getValue());
+                    if (value != null) {
+                      m.put(resourceAttributePrefix + kv.getKey(), value);
+                    }
+                  },
+                  HashMap::putAll);
           return resourceMetrics.getInstrumentationLibraryMetricsList()
               .stream()
               .flatMap(libraryMetrics -> libraryMetrics.getMetricsList()
@@ -142,8 +150,8 @@ private InputRow parseNumberDataPoint(NumberDataPoint dataPoint,
   {
 
     int capacity = resourceAttributes.size()
-            + dataPoint.getAttributesCount()
-            + 2; // metric name + value columns
+          + dataPoint.getAttributesCount()
+          + 2; // metric name + value columns
     Map<String, Object> event = Maps.newHashMapWithExpectedSize(capacity);
     event.put(metricDimension, metricName);
 
@@ -154,32 +162,34 @@ private InputRow parseNumberDataPoint(NumberDataPoint dataPoint,
     }
 
     event.putAll(resourceAttributes);
-    dataPoint.getAttributesList().forEach(att -> event.put(metricAttributePrefix + att.getKey(),
-        parseAnyValue(att.getValue())));
+    dataPoint.getAttributesList().forEach(att -> {
+      Object value = parseAnyValue(att.getValue());
+      if (value != null) {
+        event.put(metricAttributePrefix + att.getKey(), value);
+      }
+    });
 
     return createRow(TimeUnit.NANOSECONDS.toMillis(dataPoint.getTimeUnixNano()), event);
   }
 
+  @Nullable
   private static Object parseAnyValue(AnyValue value)
   {
     switch (value.getValueCase()) {
       case INT_VALUE:
         return value.getIntValue();
       case BOOL_VALUE:
         return value.getBoolValue();
-      case ARRAY_VALUE:
-        return value.getArrayValue();
-      case BYTES_VALUE:
-        return value.getBytesValue();
       case DOUBLE_VALUE:
         return value.getDoubleValue();
-      case KVLIST_VALUE:
-        return value.getKvlistValue();
       case STRING_VALUE:
         return value.getStringValue();
+
+      // TODO: Support KVLIST_VALUE, ARRAY_VALUE and BYTES_VALUE
+
       default:
-        // VALUE_NOT_SET:
-        return "";
+        // VALUE_NOT_SET
+        return null;
     }
   }
 

extensions-contrib/opentelemetry-extensions/src/main/resources/META-INF/services/org.apache.druid.initialization.DruidModule

@@ -13,4 +13,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-org.apache.druid.data.input.opencensus.protobuf.OpenCensusProtobufExtensionsModule
+org.apache.druid.data.input.opentelemetry.protobuf.OpenTelemetryMetricsProtobufInputFormat