fix(kafka): fix infinite deserialization logging

metadata-service/configuration/src/main/java/com/linkedin/metadata/config/kafka/ConsumerConfiguration.java

@@ -6,4 +6,5 @@
 public class ConsumerConfiguration {
 
   private int maxPartitionFetchBytes;
+  private boolean stopOnDeserializationError;
 }

metadata-service/configuration/src/main/resources/application.yml

@@ -236,6 +236,7 @@ kafka:
     maxRequestSize: ${KAFKA_PRODUCER_MAX_REQUEST_SIZE:5242880} # the max bytes sent by the producer, also see kafka-setup MAX_MESSAGE_BYTES for matching value
   consumer:
     maxPartitionFetchBytes: ${KAFKA_CONSUMER_MAX_PARTITION_FETCH_BYTES:5242880} # the max bytes consumed per partition
+    stopOnDeserializationError: ${KAFKA_CONSUMER_STOP_ON_DESERIALIZATION_ERROR:true} # Stops kafka listener container on deserialization error, allows user to fix problems before moving past problematic offset. If false will log and move forward past the offset
   schemaRegistry:
     type: ${SCHEMA_REGISTRY_TYPE:KAFKA} # INTERNAL or KAFKA or AWS_GLUE
     url: ${KAFKA_SCHEMAREGISTRY_URL:http://localhost:8081}

metadata-service/factories/src/main/java/com/linkedin/gms/factory/kafka/KafkaEventConsumerFactory.java

@@ -21,6 +21,11 @@
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.listener.CommonContainerStoppingErrorHandler;
+import org.springframework.kafka.listener.CommonDelegatingErrorHandler;
+import org.springframework.kafka.listener.DefaultErrorHandler;
+import org.springframework.kafka.support.serializer.DeserializationException;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
 
 @Slf4j
 @Configuration
@@ -66,8 +71,6 @@ private static Map<String, Object> buildCustomizedProperties(
       SchemaRegistryConfig schemaRegistryConfig) {
     KafkaProperties.Consumer consumerProps = baseKafkaProperties.getConsumer();
 
-    // Specify (de)serializers for record keys and for record values.
-    consumerProps.setKeyDeserializer(StringDeserializer.class);
     // Records will be flushed every 10 seconds.
     consumerProps.setEnableAutoCommit(true);
     consumerProps.setAutoCommitInterval(Duration.ofSeconds(10));
@@ -81,7 +84,13 @@ private static Map<String, Object> buildCustomizedProperties(
 
     Map<String, Object> customizedProperties = baseKafkaProperties.buildConsumerProperties();
     customizedProperties.put(
-        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, schemaRegistryConfig.getDeserializer());
+        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
+    customizedProperties.put(
+        ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class);
+    customizedProperties.put(
+        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
+    customizedProperties.put(
+        ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, schemaRegistryConfig.getDeserializer());
 
     // Override KafkaProperties with SchemaRegistryConfig only for non-empty values
     schemaRegistryConfig.getProperties().entrySet().stream()
@@ -98,14 +107,27 @@ private static Map<String, Object> buildCustomizedProperties(
   @Bean(name = "kafkaEventConsumer")
   protected KafkaListenerContainerFactory<?> createInstance(
       @Qualifier("kafkaConsumerFactory")
-          DefaultKafkaConsumerFactory<String, GenericRecord> kafkaConsumerFactory) {
+          DefaultKafkaConsumerFactory<String, GenericRecord> kafkaConsumerFactory,
+      @Qualifier("configurationProvider") ConfigurationProvider configurationProvider) {
 
     ConcurrentKafkaListenerContainerFactory<String, GenericRecord> factory =
         new ConcurrentKafkaListenerContainerFactory<>();
     factory.setConsumerFactory(kafkaConsumerFactory);
     factory.setContainerCustomizer(new ThreadPoolContainerCustomizer());
     factory.setConcurrency(kafkaEventConsumerConcurrency);
 
+    /* Sets up a delegating error handler for Deserialization errors, if disabled will
+     use DefaultErrorHandler (does back-off retry and then logs) rather than stopping the container. Stopping the container
+     prevents lost messages until the error can be examined, disabling this will allow progress, but may lose data
+    */
+    if (configurationProvider.getKafka().getConsumer().isStopOnDeserializationError()) {
+      CommonDelegatingErrorHandler delegatingErrorHandler =
+          new CommonDelegatingErrorHandler(new DefaultErrorHandler());
+      delegatingErrorHandler.addDelegate(
+          DeserializationException.class, new CommonContainerStoppingErrorHandler());
+      factory.setCommonErrorHandler(delegatingErrorHandler);
+    }
+
     log.info(
         String.format(
             "Event-based KafkaListenerContainerFactory built successfully. Consumer concurrency = %s",