feat(messaging): add transport abstractions and Kafka-only wiring (#17481)

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: david-leifker

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/config/GeneralUpgradeConfiguration.java

@@ -4,10 +4,12 @@
 import com.linkedin.gms.factory.auth.DataHubAuthorizerFactory;
 import com.linkedin.gms.factory.event.ExternalEventsServiceFactory;
 import com.linkedin.gms.factory.event.KafkaConsumerPoolFactory;
+import com.linkedin.gms.factory.event.KafkaExternalEventsPollHandlerConfiguration;
 import com.linkedin.gms.factory.graphql.GraphQLEngineFactory;
 import com.linkedin.gms.factory.kafka.KafkaEventConsumerFactory;
 import com.linkedin.gms.factory.kafka.SimpleKafkaConsumerFactory;
 import com.linkedin.gms.factory.kafka.trace.KafkaTraceReaderFactory;
+import com.linkedin.gms.factory.messaging.KafkaConsumerLagPort;
 import com.linkedin.gms.factory.telemetry.ScheduledAnalyticsFactory;
 import com.linkedin.gms.factory.trace.TraceServiceFactory;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -18,6 +20,9 @@
 /**
  * Configuration for general upgrades that includes most components but excludes some that are not
  * typically needed for upgrade operations.
+ *
+ * <p>Consumer lag ports and their trace-reader dependencies are excluded because the system-update
+ * context excludes {@link KafkaTraceReaderFactory} and {@link TraceServiceFactory}.
  */
 @Configuration
 @EnableAutoConfiguration
@@ -41,7 +46,9 @@
             KafkaTraceReaderFactory.class,
             TraceServiceFactory.class,
             KafkaConsumerPoolFactory.class,
-            ExternalEventsServiceFactory.class
+            KafkaExternalEventsPollHandlerConfiguration.class,
+            ExternalEventsServiceFactory.class,
+            KafkaConsumerLagPort.class
           })
     })
 public class GeneralUpgradeConfiguration {}

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/config/KafkaSetupConfig.java

@@ -4,6 +4,7 @@
 import com.linkedin.datahub.upgrade.system.BlockingSystemUpgrade;
 import com.linkedin.datahub.upgrade.system.kafka.KafkaSetup;
 import com.linkedin.gms.factory.config.ConfigurationProvider;
+import com.linkedin.metadata.config.messaging.KafkaMessagingEnabledCondition;
 import io.datahubproject.metadata.context.OperationContext;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
@@ -13,13 +14,14 @@
 import org.springframework.core.annotation.Order;
 
 @Configuration
-@Conditional(SystemUpdateCondition.BlockingSystemUpdateCondition.class)
+@Conditional(KafkaMessagingEnabledCondition.class)
 public class KafkaSetupConfig {
 
   @Autowired private OperationContext opContext;
 
-  @Order(1) // This ensures it runs before BuildIndices (@Order(2))
+  @Order(1) // Before pgSearch entity schema (when enabled) and BuildIndices
   @Bean(name = "kafkaSetup")
+  @Conditional(SystemUpdateCondition.BlockingSystemUpdateCondition.class)
   public BlockingSystemUpgrade kafkaSetup(
       final ConfigurationProvider configurationProvider, KafkaProperties properties) {
     return new KafkaSetup(opContext, configurationProvider.getKafka(), properties);

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/config/OpenTelemetryConfig.java

@@ -2,7 +2,7 @@
 
 import com.linkedin.gms.factory.config.ConfigurationProvider;
 import com.linkedin.gms.factory.system_telemetry.OpenTelemetryBaseFactory;
-import com.linkedin.metadata.event.GenericProducer;
+import com.linkedin.metadata.event.UsageEventPublisher;
 import com.linkedin.metadata.utils.metrics.MetricUtils;
 import io.datahubproject.metadata.context.SystemTelemetryContext;
 import javax.annotation.Nullable;
@@ -25,7 +25,7 @@ protected SystemTelemetryContext traceContext(
       MetricUtils metricUtils,
       ConfigurationProvider configurationProvider,
       @Autowired(required = false) @Qualifier("dataHubUsageEventProducer") @Nullable
-          GenericProducer<String> dueProducer) {
-    return super.traceContext(metricUtils, configurationProvider, dueProducer);
+          UsageEventPublisher usageEventPublisher) {
+    return super.traceContext(metricUtils, configurationProvider, usageEventPublisher);
   }
 }

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/config/SystemUpdateConfig.java

@@ -12,14 +12,10 @@
 import com.linkedin.entity.client.EntityClientConfig;
 import com.linkedin.entity.client.SystemEntityClient;
 import com.linkedin.gms.factory.config.ConfigurationProvider;
-import com.linkedin.gms.factory.kafka.DataHubKafkaProducerFactory;
-import com.linkedin.gms.factory.kafka.common.TopicConventionFactory;
 import com.linkedin.gms.factory.kafka.schemaregistry.InternalSchemaRegistryFactory;
 import com.linkedin.metadata.client.SystemJavaEntityClient;
 import com.linkedin.metadata.config.cache.client.EntityClientCacheConfig;
 import com.linkedin.metadata.config.kafka.KafkaConfiguration;
-import com.linkedin.metadata.dao.producer.KafkaEventProducer;
-import com.linkedin.metadata.dao.producer.KafkaHealthChecker;
 import com.linkedin.metadata.dao.throttle.ThrottleSensor;
 import com.linkedin.metadata.entity.AspectDao;
 import com.linkedin.metadata.entity.DeleteEntityService;
@@ -35,7 +31,6 @@
 import com.linkedin.metadata.timeseries.TimeseriesAspectService;
 import com.linkedin.metadata.utils.metrics.MetricUtils;
 import com.linkedin.metadata.version.GitVersion;
-import com.linkedin.mxe.TopicConvention;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
@@ -44,14 +39,9 @@
 import javax.annotation.Nonnull;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.avro.generic.IndexedRecord;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.Producer;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
@@ -142,52 +132,13 @@ public String getRevision() {
 
   @Bean
   public DataHubStartupStep dataHubStartupStep(
-      @Qualifier("duheKafkaEventProducer") final KafkaEventProducer kafkaEventProducer,
+      @Qualifier("duheKafkaEventProducer") final EventProducer kafkaEventProducer,
       final GitVersion gitVersion,
       @Qualifier("revision") String revision) {
     return new DataHubStartupStep(
         kafkaEventProducer, String.format("%s-%s", gitVersion.getVersion(), revision));
   }
 
-  @Autowired
-  @Qualifier(TopicConventionFactory.TOPIC_CONVENTION_BEAN)
-  private TopicConvention topicConvention;
-
-  @Autowired private KafkaHealthChecker kafkaHealthChecker;
-
-  @Bean(name = "duheKafkaEventProducer")
-  protected KafkaEventProducer duheKafkaEventProducer(
-      @Qualifier("configurationProvider") ConfigurationProvider provider,
-      KafkaProperties properties,
-      @Qualifier("duheSchemaRegistryConfig")
-          KafkaConfiguration.SerDeKeyValueConfig duheSchemaRegistryConfig,
-      MetricUtils metricUtils) {
-    KafkaConfiguration kafkaConfiguration = provider.getKafka();
-    Producer<String, IndexedRecord> producer =
-        new KafkaProducer<>(
-            DataHubKafkaProducerFactory.buildProducerProperties(
-                duheSchemaRegistryConfig, kafkaConfiguration, properties));
-    return new KafkaEventProducer(producer, topicConvention, kafkaHealthChecker, metricUtils);
-  }
-
-  /**
-   * The ReindexDataJobViaNodesCLLConfig step requires publishing to MCL. Overriding the default
-   * producer with this special producer which doesn't require an active registry.
-   *
-   * <p>Use when INTERNAL registry and is SYSTEM_UPDATE
-   *
-   * <p>This forces this producer into the EntityService
-   */
-  @Primary
-  @Bean(name = "kafkaEventProducer")
-  @ConditionalOnProperty(
-      name = "kafka.schemaRegistry.type",
-      havingValue = InternalSchemaRegistryFactory.TYPE)
-  protected KafkaEventProducer kafkaEventProducer(
-      @Qualifier("duheKafkaEventProducer") KafkaEventProducer kafkaEventProducer) {
-    return kafkaEventProducer;
-  }
-
   @Primary
   @Bean(name = "schemaRegistryConfig")
   @ConditionalOnProperty(
@@ -208,7 +159,7 @@ protected KafkaConfiguration.SerDeKeyValueConfig schemaRegistryConfig(
   @Conditional(SystemUpdateCondition.BlockingSystemUpdateCondition.class)
   @Nonnull
   protected EntityService<ChangeItemImpl> createEntityServiceWithSystemUpdateCDCMode(
-      @Qualifier("kafkaEventProducer") final KafkaEventProducer eventProducer,
+      @Qualifier("kafkaEventProducer") final EventProducer eventProducer,
       @Qualifier("entityAspectDao") final AspectDao aspectDao,
       @Qualifier("configurationProvider") ConfigurationProvider configurationProvider,
       @Value("${featureFlags.showBrowseV2}") final boolean enableBrowsePathV2,

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/config/SystemUpdateKafkaMessagingConfig.java

@@ -0,0 +1,69 @@
+package com.linkedin.datahub.upgrade.config;
+
+import com.linkedin.gms.factory.config.ConfigurationProvider;
+import com.linkedin.gms.factory.kafka.DataHubKafkaProducerFactory;
+import com.linkedin.gms.factory.kafka.common.TopicConventionFactory;
+import com.linkedin.gms.factory.kafka.schemaregistry.InternalSchemaRegistryFactory;
+import com.linkedin.metadata.config.kafka.KafkaConfiguration;
+import com.linkedin.metadata.config.messaging.KafkaMessagingEnabledCondition;
+import com.linkedin.metadata.dao.producer.KafkaEventProducer;
+import com.linkedin.metadata.dao.producer.KafkaHealthChecker;
+import com.linkedin.metadata.utils.metrics.MetricUtils;
+import com.linkedin.mxe.TopicConvention;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+
+@Slf4j
+@Configuration
+@Conditional(KafkaMessagingEnabledCondition.class)
+public class SystemUpdateKafkaMessagingConfig {
+
+  @Autowired
+  @Qualifier(TopicConventionFactory.TOPIC_CONVENTION_BEAN)
+  private TopicConvention topicConvention;
+
+  @Autowired private KafkaHealthChecker kafkaHealthChecker;
+
+  @Bean(name = "duheKafkaEventProducer")
+  protected KafkaEventProducer duheKafkaEventProducer(
+      @Qualifier("configurationProvider") ConfigurationProvider provider,
+      KafkaProperties properties,
+      @Qualifier("duheSchemaRegistryConfig")
+          KafkaConfiguration.SerDeKeyValueConfig duheSchemaRegistryConfig,
+      MetricUtils metricUtils) {
+    KafkaConfiguration kafkaConfiguration = provider.getKafka();
+    Producer<String, IndexedRecord> producer =
+        new KafkaProducer<>(
+            DataHubKafkaProducerFactory.buildProducerProperties(
+                duheSchemaRegistryConfig, kafkaConfiguration, properties));
+    return new KafkaEventProducer(producer, topicConvention, kafkaHealthChecker, metricUtils);
+  }
+
+  /**
+   * The ReindexDataJobViaNodesCLLConfig step requires publishing to MCL. Overriding the default
+   * producer with this special producer which doesn't require an active registry.
+   *
+   * <p>Use when INTERNAL registry and is SYSTEM_UPDATE
+   *
+   * <p>This forces this producer into the EntityService
+   */
+  @Primary
+  @Bean(name = "kafkaEventProducer")
+  @ConditionalOnProperty(
+      name = "kafka.schemaRegistry.type",
+      havingValue = InternalSchemaRegistryFactory.TYPE)
+  protected KafkaEventProducer kafkaEventProducer(
+      @Qualifier("duheKafkaEventProducer") KafkaEventProducer kafkaEventProducer) {
+    return kafkaEventProducer;
+  }
+}

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/system/elasticsearch/steps/DataHubStartupStep.java

@@ -4,7 +4,7 @@
 import com.linkedin.datahub.upgrade.UpgradeStep;
 import com.linkedin.datahub.upgrade.UpgradeStepResult;
 import com.linkedin.datahub.upgrade.impl.DefaultUpgradeStepResult;
-import com.linkedin.metadata.dao.producer.KafkaEventProducer;
+import com.linkedin.metadata.event.EventProducer;
 import com.linkedin.mxe.DataHubUpgradeHistoryEvent;
 import com.linkedin.upgrade.DataHubUpgradeState;
 import java.util.function.Function;
@@ -14,7 +14,7 @@
 @RequiredArgsConstructor
 @Slf4j
 public class DataHubStartupStep implements UpgradeStep {
-  private final KafkaEventProducer _kafkaEventProducer;
+  private final EventProducer _kafkaEventProducer;
   private final String _version;
 
   @Override