fix(ingestion): handle LossyList sentinel string in LogEntry deserialization

[Dutt23]

Summary

• The Python ingestion framework's LossyList appends a plain-string sentinel (e.g. "... sampled of N total elements") as the 11th element when a warnings/failures/infos list exceeds 10 entries.

• Jackson throws MismatchedInputException when it encounters that sentinel string while deserializing List<LogEntry>, causing IngestionMetricsEmitter to crash the Kafka consumer thread for high-failure runs.

• Fix: add LogEntryDeserializer that maps string tokens to an empty LogEntry (which toLogEntryMaps() silently drops) and delegates object tokens to standard bean deserialization; annotate all four List<LogEntry> fields with @JsonDeserialize(contentUsing = LogEntryDeserializer.class).

• Couple of failures . https://github.com/datahub-project/datahub/actions/runs/27151079465/job/80151440137?pr=17808
  https://github.com/datahub-project/datahub/actions/runs/27152353194/job/80153609414

• 2026-06-08 17:11:42,178 [ThreadPoolTaskExecutor-1] ERROR c.l.m.i.IngestionMetricsEmitter urn=none aspect=none entityType=none changeType=none - Error recording metrics for urn:li:dataHubExecutionRequest:datapack-showcase-ecommerce-1780938689918: Cannot construct instance of com.linkedin.metadata.ingestion.IngestionRunReport$LogEntry (although at least one Creator exists): no String-argument constructor/factory method to deserialize from String value ('... sampled of 1246 total elements') at [Source: REDACTED (StreamReadFeature.INCLUDE_SOURCE_IN_LOCATIONdisabled); line: 115, column: 9] (through reference chain: com.linkedin.metadata.ingestion.IngestionRunReport$Report["source"]->com.linkedin.metadata.ingestion.IngestionRunReport$SourceSection["report"]->com.linkedin.metadata.ingestion.IngestionRunReport$SourceReport["failures"]->java.util.ArrayList[10]) com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot construct instance ofcom.linkedin.metadata.ingestion.IngestionRunReport$LogEntry (although at least one Creator exists): no String-argument constructor/factory method to deserialize from String value ('... sampled of 1246 total elements') at [Source: REDACTED (StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION disabled); line: 115, column: 9] (through reference chain: com.linkedin.metadata.ingestion.IngestionRunReport$Report["source"]->com.linkedin.metadata.ingestion.IngestionRunReport$SourceSection["report"]->com.linkedin.metadata.ingestion.IngestionRunReport$SourceReport["failures"]->java.util.ArrayList[10]) at com.fasterxml.jackson.databind.exc.MismatchedInputException.from(MismatchedInputException.java:76) at com.fasterxml.jackson.databind.DeserializationContext.reportInputMismatch(DeserializationContext.java:1801) at com.fasterxml.jackson.databind.DeserializationContext.handleMissingInstantiator(DeserializationContext.java:1426) at com.fasterxml.jackson.databind.deser.std.StdDeserializer._deserializeFromString(StdDeserializer.java:310) at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.deserializeFromString(BeanDeserializerBase.java:1594) at com.fasterxml.jackson.databind.deser.BeanDeserializer._deserializeOther(BeanDeserializer.java:189) at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:179) at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer._deserializeNoNullChecks(CollectionDeserializer.java:545) at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer._deserializeFromArray(CollectionDeserializer.java:358) at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer.deserialize(CollectionDeserializer.java:245) at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer.deserialize(CollectionDeserializer.java:29) at com.fasterxml.jackson.databind.deser.impl.MethodProperty.deserializeAndSet(MethodProperty.java:129) at com.fasterxml.jackson.databind.deser.BeanDeserializer.vanillaDeserialize(BeanDeserializer.java:302) at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:169) at com.fasterxml.jackson.databind.deser.impl.MethodProperty.deserializeAndSet(MethodProperty.java:129) at com.fasterxml.jackson.databind.deser.BeanDeserializer.vanillaDeserialize(BeanDeserializer.java:302) at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:169) at com.fasterxml.jackson.databind.deser.impl.MethodProperty.deserializeAndSet(MethodProperty.java:129) at com.fasterxml.jackson.databind.deser.BeanDeserializer.vanillaDeserialize(BeanDeserializer.java:302) at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:169) at com.fasterxml.jackson.databind.deser.DefaultDeserializationContext.readRootValue(DefaultDeserializationContext.java:342) at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:5052) at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3954) at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3922) at com.linkedin.metadata.ingestion.IngestionMetricsEmitter.recordMetrics(IngestionMetricsEmitter.java:125) at com.linkedin.metadata.ingestion.IngestionMetricsEmitter.observeMCPs(IngestionMetricsEmitter.java:96) at com.linkedin.metadata.aspect.plugins.hooks.MCPObserver.apply(MCPObserver.java:29) at com.linkedin.metadata.aspect.batch.AspectsBatch.applyMCPObservers(AspectsBatch.java:215) at com.linkedin.metadata.aspect.batch.AspectsBatch.applyMCPObservers(AspectsBatch.java:207) at com.linkedin.metadata.entity.EntityServiceImpl.ingestProposal(EntityServiceImpl.java:1693) at com.linkedin.metadata.client.JavaEntityClient.lambda$batchIngestProposals$11(JavaEntityClient.java:784) at java.base/java.util.Iterator.forEachRemaining(Unknown Source) at com.linkedin.metadata.client.JavaEntityClient.batchIngestProposals(JavaEntityClient.java:772) at com.linkedin.metadata.kafka.batch.BatchMetadataChangeProposalsProcessor.processBatch(BatchMetadataChangeProposalsProcessor.java:214) at com.linkedin.metadata.kafka.batch.BatchMetadataChangeProposalsProcessor.processInBatches(BatchMetadataChangeProposalsProcessor.java:188) at com.linkedin.metadata.kafka.batch.BatchMetadataChangeProposalsProcessor.lambda$consume$1(BatchMetadataChangeProposalsProcessor.java:140) at io.datahubproject.metadata.context.SystemTelemetryContext.withQueueSpan(SystemTelemetryContext.java:230) at io.datahubproject.metadata.context.OperationContext.withQueueSpan(OperationContext.java:455) at com.linkedin.metadata.kafka.batch.BatchMetadataChangeProposalsProcessor.consume(BatchMetadataChangeProposalsProcessor.java:133) at com.linkedin.metadata.kafka.batch.BatchMetadataChangeProposalsProcessor.consume(BatchMetadataChangeProposalsProcessor.java:72) at com.linkedin.metadata.kafka.batch.BatchMetadataChangeProposalsKafkaListener.consume(BatchMetadataChangeProposalsKafkaListener.java:38) at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(Unknown Source) at java.base/java.lang.reflect.Method.invoke(Unknown Source) at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:168) at org.springframework.kafka.listener.adapter.KotlinAwareInvocableHandlerMethod.doInvoke(KotlinAwareInvocableHandlerMethod.java:48) at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:119) at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:80) at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:488) at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invoke(MessagingMessageListenerAdapter.java:425) at org.springframework.kafka.listener.adapter.BatchMessagingMessageListenerAdapter.onMessage(BatchMessagingMessageListenerAdapter.java:194) at org.springframework.kafka.listener.adapter.BatchMessagingMessageListenerAdapter.onMessage(BatchMessagingMessageListenerAdapter.java:62) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeBatchOnMessage(KafkaMessageListenerContainer.java:2522) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeBatchOnMessageWithRecordsOrList(KafkaMessageListenerContainer.java:2503) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeBatchOnMessage(KafkaMessageListenerContainer.java:2420) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeBatchListener(KafkaMessageListenerContainer.java:2298) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeBatchListener(KafkaMessageListenerContainer.java:2222) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:2201) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1559) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1499) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1368) at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(Unknown Source) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) at java.base/java.lang.Thread.run(Unknown Source)

Changes

• IngestionRunReport.java: add Jackson imports, @JsonDeserialize(contentUsing = LogEntryDeserializer.class) on SourceReport.warnings, SourceReport.failures, SourceReport.infos, and SinkReport.failures; add new inner class LogEntryDeserializer.
• IngestionMetricsEmitter.java: update NOTE comment to accurately describe the LossyList sentinel handling; add null-entry guard in toLogEntryMaps(); update Javadoc for that method.
• IngestionMetricsEmitterTest.java: add regression test testLossyListSentinelDoesNotThrow that passes a report with mixed object + string sentinel elements and verifies correct counter values.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ All tests successful. No failed tests found.

📢 Thoughts on this report? Let us know!