telemetry: add support for unified mixpanel + kafka tracking in GMS (#5514)

Author: shirshanka

.github/workflows/docker-unified.yml

@@ -365,6 +365,10 @@ jobs:
       fail-fast: false
       matrix: ${{ fromJson(needs.smoke_test_matrix.outputs.matrix) }}
     if: ${{ always() && !failure() && !cancelled() && needs.smoke_test_matrix.outputs.matrix != '[]' }}
+    env:
+      # TODO Chakru: Review if required
+      MIXPANEL_API_SECRET: ${{ secrets.MIXPANEL_API_SECRET }}
+      MIXPANEL_PROJECT_ID: ${{ secrets.MIXPANEL_PROJECT_ID }}
     steps:
       - name: Free up disk space
         if: ${{ !contains(needs.setup.outputs.test_runner_type, 'depot') }}

metadata-events/mxe-registration/src/main/java/com/linkedin/mxe/TopicConvention.java

@@ -57,6 +57,10 @@ public interface TopicConvention {
   @Nonnull
   String getPlatformEventTopicName();
 
+  /** The name of the datahub usage event topic. */
+  @Nonnull
+  String getDataHubUsageEventTopicName();
+
   /**
    * Returns the name of the metadata change event (v5) kafka topic.
    *

metadata-events/mxe-registration/src/main/java/com/linkedin/mxe/TopicConventionImpl.java

@@ -41,7 +41,7 @@ public final class TopicConventionImpl implements TopicConvention {
   private final String _failedMetadataChangeProposalTopicName;
   private final String _platformEventTopicName;
   private final String _dataHubUpgradeHistoryTopicName;
-
+  private final String _dataHubUsageEventTopicName;
   // v5 patterns
   private final String _eventPattern;
 
@@ -55,7 +55,8 @@ public TopicConventionImpl(
       @Nonnull String failedMetadataChangeProposalTopicName,
       @Nonnull String platformEventTopicName,
       @Nonnull String eventPattern,
-      @Nonnull String dataHubUpgradeHistoryTopicName) {
+      @Nonnull String dataHubUpgradeHistoryTopicName,
+      @Nonnull String dataHubUsageEventTopicName) {
     _metadataChangeEventTopicName = metadataChangeEventTopicName;
     _metadataAuditEventTopicName = metadataAuditEventTopicName;
     _failedMetadataChangeEventTopicName = failedMetadataChangeEventTopicName;
@@ -66,6 +67,7 @@ public TopicConventionImpl(
     _platformEventTopicName = platformEventTopicName;
     _eventPattern = eventPattern;
     _dataHubUpgradeHistoryTopicName = dataHubUpgradeHistoryTopicName;
+    _dataHubUsageEventTopicName = dataHubUsageEventTopicName;
   }
 
   public TopicConventionImpl() {
@@ -79,7 +81,8 @@ public TopicConventionImpl() {
         Topics.FAILED_METADATA_CHANGE_PROPOSAL,
         Topics.PLATFORM_EVENT,
         DEFAULT_EVENT_PATTERN,
-        Topics.DATAHUB_UPGRADE_HISTORY_TOPIC_NAME);
+        Topics.DATAHUB_UPGRADE_HISTORY_TOPIC_NAME,
+        Topics.DATAHUB_USAGE_EVENT);
   }
 
   @Nonnull
@@ -130,6 +133,12 @@ public String getPlatformEventTopicName() {
     return _platformEventTopicName;
   }
 
+  @Nonnull
+  @Override
+  public String getDataHubUsageEventTopicName() {
+    return _dataHubUsageEventTopicName;
+  }
+
   @Nonnull
   private String buildEventName(
       @Nonnull String eventType,

metadata-jobs/mae-consumer/src/main/java/com/linkedin/metadata/kafka/transformer/DataHubUsageEventTransformer.java

@@ -77,21 +77,25 @@ public Optional<TransformedDocument> transformDataHubUsageEvent(String dataHubUs
     // Timestamp is required
     long timestampMillis;
     if (usageEvent.get(TIMESTAMP).isNumber()) {
+      // Handle numeric timestamp (epoch milliseconds)
       timestampMillis = usageEvent.get(TIMESTAMP).asLong();
     } else if (usageEvent.get(TIMESTAMP).isTextual()) {
+      // Handle ISO date string
       try {
         String isoDate = usageEvent.get(TIMESTAMP).asText();
         java.time.Instant instant = java.time.Instant.parse(isoDate);
         timestampMillis = instant.toEpochMilli();
       } catch (Exception e) {
-        log.warn("Failed to parse ISO date string: {}", usageEvent.get(TIMESTAMP));
+        log.warn(
+            "Failed to parse ISO date string: {}. Defaulting to current system time",
+            usageEvent.get(TIMESTAMP));
         timestampMillis = Instant.now().toEpochMilli();
       }
     } else {
       log.warn(
           "Invalid timestamp format - expected number or ISO date string but got: {}",
           usageEvent.get(TIMESTAMP));
-      timestampMillis = Instant.now().toEpochMilli();
+      return Optional.empty();
     }
 
     log.debug("Raw timestamp value from event: {}", timestampMillis);
@@ -108,11 +112,11 @@ public Optional<TransformedDocument> transformDataHubUsageEvent(String dataHubUs
     }
 
     try {
-      return Optional.of(
-          new TransformedDocument(
-              getId(eventDocument), OBJECT_MAPPER.writeValueAsString(eventDocument)));
+      String serializedDoc = OBJECT_MAPPER.writeValueAsString(eventDocument);
+      log.debug("Final serialized document: {}", serializedDoc);
+      return Optional.of(new TransformedDocument(getId(eventDocument), serializedDoc));
     } catch (JsonProcessingException e) {
-      log.info("Failed to package document: {}", eventDocument);
+      log.error("Failed to package document: {}", eventDocument);
       return Optional.empty();
     }
   }

metadata-service/auth-impl/src/main/java/com/datahub/telemetry/TrackingDestination.java

@@ -0,0 +1,7 @@
+package com.datahub.telemetry;
+
+/** Enum representing the possible destinations for tracking events. */
+public enum TrackingDestination {
+  MIXPANEL,
+  KAFKA
+}