fix? status type can be null too?

airbytehq · Nov 25, 2024 · 114a17d · 114a17d
1 parent 59807a1
commit 114a17d
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 1 deletion.
diff --git a/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/command/DestinationCatalog.kt b/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/command/DestinationCatalog.kt
@@ -5,6 +5,7 @@
 package io.airbyte.cdk.load.command
 
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog
+import io.github.oshai.kotlinlogging.KotlinLogging
 import io.micronaut.context.annotation.Factory
 import io.micronaut.context.annotation.Secondary
 import jakarta.inject.Singleton
@@ -14,6 +15,8 @@ import jakarta.inject.Singleton
  * for usability.
  */
 data class DestinationCatalog(val streams: List<DestinationStream> = emptyList()) {
+    private val log = KotlinLogging.logger {}
+
     private val byDescriptor: Map<DestinationStream.Descriptor, DestinationStream> =
         streams.associateBy { it.descriptor }
 
@@ -23,6 +26,7 @@ data class DestinationCatalog(val streams: List<DestinationStream> = emptyList()
                 "Catalog must have at least one stream: check that files are in the correct location."
             )
         }
+        log.info { "Destination catalog initialized: $streams"}
     }
 
     fun getStream(name: String, namespace: String?): DestinationStream {

diff --git a/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/message/DestinationMessage.kt b/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/message/DestinationMessage.kt
@@ -435,7 +435,7 @@ class DestinationMessageFactory(
                         namespace = status.streamDescriptor.namespace,
                         name = status.streamDescriptor.name,
                     )
-                if (message.trace.type == AirbyteTraceMessage.Type.STREAM_STATUS) {
+                if ((message.trace.type ?: AirbyteTraceMessage.Type.STREAM_STATUS) == AirbyteTraceMessage.Type.STREAM_STATUS) {
                     when (status.status) {
                         AirbyteStreamStatus.COMPLETE ->
                             if (fileTransferEnabled) {