chore: hashing mapper populate meta.changes on errors (#13919)

Author: gosusnp

airbyte-config/config-models/src/main/kotlin/io/airbyte/config/adapters/AirbyteRecord.kt

@@ -4,6 +4,15 @@ import io.airbyte.config.StreamDescriptor
 import io.airbyte.protocol.models.AirbyteMessage
 
 interface AirbyteRecord {
+  enum class Change {
+    NULLED,
+    TRUNCATED,
+  }
+
+  enum class Reason {
+    PLATFORM_SERIALIZATION_ERROR,
+  }
+
   val streamDescriptor: StreamDescriptor
   val asProtocol: AirbyteMessage
 
@@ -17,6 +26,12 @@ interface AirbyteRecord {
     fieldName: String,
     value: T,
   )
+
+  fun trackFieldError(
+    fieldName: String,
+    change: Change,
+    reason: Reason,
+  )
 }
 
 interface Value {

airbyte-config/config-models/src/main/kotlin/io/airbyte/config/adapters/JsonAdapters.kt

@@ -8,6 +8,8 @@ import com.fasterxml.jackson.databind.node.ObjectNode
 import com.fasterxml.jackson.databind.node.TextNode
 import io.airbyte.config.StreamDescriptor
 import io.airbyte.protocol.models.AirbyteMessage
+import io.airbyte.protocol.models.AirbyteRecordMessageMeta
+import io.airbyte.protocol.models.AirbyteRecordMessageMetaChange
 
 class JsonValueAdapter(private val node: JsonNode) : Value {
   override fun asBoolean(): Boolean = node.asBoolean()
@@ -38,6 +40,30 @@ data class AirbyteJsonRecordAdapter(private val message: AirbyteMessage) : Airby
     data.set<JsonNode>(fieldName, createNode(value))
   }
 
+  override fun trackFieldError(
+    fieldName: String,
+    change: AirbyteRecord.Change,
+    reason: AirbyteRecord.Reason,
+  ) {
+    val metaChange =
+      AirbyteRecordMessageMetaChange()
+        .withChange(change.toProtocol())
+        .withField(fieldName)
+        .withReason(reason.toProtocol())
+
+    // handling all the cascading layers of potential null objects
+    // very thread-unsafe
+    if (message.record != null) {
+      if (message.record.meta == null) {
+        message.record.withMeta(AirbyteRecordMessageMeta().withChanges(mutableListOf()))
+      }
+      if (message.record.meta.changes == null) {
+        message.record.meta.withChanges(mutableListOf())
+      }
+      message.record.meta.changes.add(metaChange)
+    }
+  }
+
   private fun <T : Any> createNode(value: T): JsonNode =
     when (value) {
       is Boolean -> BooleanNode.valueOf(value)
@@ -46,4 +72,15 @@ data class AirbyteJsonRecordAdapter(private val message: AirbyteMessage) : Airby
       is String -> TextNode.valueOf(value)
       else -> TODO("Unsupported type ${value::class.java.name}")
     }
+
+  private fun AirbyteRecord.Change.toProtocol() =
+    when (this) {
+      AirbyteRecord.Change.NULLED -> AirbyteRecordMessageMetaChange.Change.NULLED
+      AirbyteRecord.Change.TRUNCATED -> AirbyteRecordMessageMetaChange.Change.TRUNCATED
+    }
+
+  private fun AirbyteRecord.Reason.toProtocol() =
+    when (this) {
+      AirbyteRecord.Reason.PLATFORM_SERIALIZATION_ERROR -> AirbyteRecordMessageMetaChange.Reason.PLATFORM_SERIALIZATION_ERROR
+    }
 }

airbyte-config/config-models/src/test/kotlin/io/airbyte/config/adapters/JsonRecordAdapterTest.kt

@@ -3,6 +3,7 @@ package io.airbyte.config.adapters
 import io.airbyte.commons.json.Jsons
 import io.airbyte.config.StreamDescriptor
 import io.airbyte.protocol.models.AirbyteMessage
+import io.airbyte.protocol.models.AirbyteRecordMessageMetaChange
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 
@@ -49,6 +50,20 @@ class JsonRecordAdapterTest {
     assertEquals("4.2", adapter.get(NUMBER_FIELD).asString())
   }
 
+  @Test
+  fun `tracking meta changes`() {
+    val adapter = getAdapterFromRecord(jsonRecordString)
+
+    adapter.trackFieldError("test", AirbyteRecord.Change.TRUNCATED, AirbyteRecord.Reason.PLATFORM_SERIALIZATION_ERROR)
+
+    val expectedMetaChange =
+      AirbyteRecordMessageMetaChange()
+        .withField("test")
+        .withChange(AirbyteRecordMessageMetaChange.Change.TRUNCATED)
+        .withReason(AirbyteRecordMessageMetaChange.Reason.PLATFORM_SERIALIZATION_ERROR)
+    assertEquals(listOf(expectedMetaChange), adapter.asProtocol.record.meta.changes)
+  }
+
   @Test
   fun `verify modify then serialize`() {
     val adapter = getAdapterFromRecord(jsonRecordString)

airbyte-mappers/src/main/kotlin/io/airbyte/mappers/transformations/HashingMapper.kt

@@ -102,13 +102,20 @@ class HashingMapper : Mapper {
     record: AirbyteRecord,
   ) {
     val (targetField, method, fieldNameSuffix) = getConfigValues(config.config)
+    val outputFieldName = "$targetField$fieldNameSuffix"
 
     if (record.has(targetField)) {
-      val data = record.get(targetField).asString().toByteArray()
-
-      val hashedAndEncodeValue: String = hashAndEncodeData(method, data)
-      record.set(targetField + fieldNameSuffix, hashedAndEncodeValue)
-      record.remove(targetField)
+      try {
+        val data = record.get(targetField).asString().toByteArray()
+
+        val hashedAndEncodeValue: String = hashAndEncodeData(method, data)
+        record.set(outputFieldName, hashedAndEncodeValue)
+      } catch (e: Exception) {
+        // TODO We should use a more precise Reason once available in the protocol
+        record.trackFieldError(outputFieldName, AirbyteRecord.Change.NULLED, AirbyteRecord.Reason.PLATFORM_SERIALIZATION_ERROR)
+      } finally {
+        record.remove(targetField)
+      }
     }
   }
 

airbyte-mappers/src/test/kotlin/io/airbyte/config/adapters/TestRecordAdapter.kt

@@ -12,7 +12,13 @@ class TestValueAdapter(private val value: Any) : Value {
 }
 
 class TestRecordAdapter(override val streamDescriptor: StreamDescriptor, data: Map<String, Any>) : AirbyteRecord {
+  data class Change(val fieldName: String, val change: AirbyteRecord.Change, val reason: AirbyteRecord.Reason)
+
   private val data: MutableMap<String, Any> = data.toMutableMap()
+  private val _changes: MutableList<Change> = mutableListOf()
+
+  val changes: List<Change>
+    get(): List<Change> = _changes.toList()
 
   override val asProtocol: AirbyteMessage
     get() = TODO("Not yet implemented")
@@ -31,4 +37,12 @@ class TestRecordAdapter(override val streamDescriptor: StreamDescriptor, data: M
   ) {
     data[fieldName] = value
   }
+
+  override fun trackFieldError(
+    fieldName: String,
+    change: AirbyteRecord.Change,
+    reason: AirbyteRecord.Reason,
+  ) {
+    _changes.add(Change(fieldName, change, reason))
+  }
 }

airbyte-mappers/src/test/kotlin/io/airbyte/mappers/transformations/HashingMapperTest.kt

@@ -4,6 +4,7 @@ import io.airbyte.config.ConfiguredMapper
 import io.airbyte.config.Field
 import io.airbyte.config.FieldType
 import io.airbyte.config.StreamDescriptor
+import io.airbyte.config.adapters.AirbyteRecord
 import io.airbyte.config.adapters.TestRecordAdapter
 import io.airbyte.mappers.transformations.HashingMapper.Companion.supportedMethods
 import io.mockk.spyk
@@ -113,9 +114,15 @@ class HashingMapperTest {
         ),
       )
 
-    assertThrows(IllegalArgumentException::class.java) {
-      hashingMapper.map(config, TestRecordAdapter(StreamDescriptor().withName("any"), mapOf("field1" to "anything")))
-    }
+    val record = TestRecordAdapter(StreamDescriptor().withName("any"), mapOf("field1" to "anything"))
+    hashingMapper.map(config, record)
+
+    assertFalse(record.has("field1"))
+    assertFalse(record.has("field1_hashed"))
+    assertEquals(
+      listOf(TestRecordAdapter.Change("field1_hashed", AirbyteRecord.Change.NULLED, AirbyteRecord.Reason.PLATFORM_SERIALIZATION_ERROR)),
+      record.changes,
+    )
   }
 
   @Test