pass streams to debezium sources on cold start (#55734)

Author: mwbayley

airbyte-cdk/bulk/toolkits/extract-cdc/src/main/kotlin/io/airbyte/cdk/read/cdc/CdcPartitionsCreator.kt

@@ -45,6 +45,7 @@ class CdcPartitionsCreator<T : Comparable<T>>(
                 "Triggering reset. Incumbent CDC state is invalid, reason: ${reason}."
             )
         }
+        var allStreams: List<Stream> = feedBootstrap.feed.streams
         val activeStreams: List<Stream> by lazy {
             feedBootstrap.feed.streams.filter { feedBootstrap.currentState(it) != null }
         }
@@ -68,7 +69,7 @@ class CdcPartitionsCreator<T : Comparable<T>>(
         val startingSchemaHistory: DebeziumSchemaHistory?
         when (warmStartState) {
             null -> {
-                debeziumProperties = creatorOps.generateColdStartProperties()
+                debeziumProperties = creatorOps.generateColdStartProperties(allStreams)
                 startingOffset = syntheticOffset
                 startingSchemaHistory = null
             }
@@ -97,7 +98,7 @@ class CdcPartitionsCreator<T : Comparable<T>>(
                 resetReason.set(warmStartState.reason)
                 log.info { "Resetting invalid incumbent CDC state with synthetic state." }
                 feedBootstrap.resetAll()
-                debeziumProperties = creatorOps.generateColdStartProperties()
+                debeziumProperties = creatorOps.generateColdStartProperties(allStreams)
                 startingOffset = syntheticOffset
                 startingSchemaHistory = null
             }

airbyte-cdk/bulk/toolkits/extract-cdc/src/main/kotlin/io/airbyte/cdk/read/cdc/DebeziumOperations.kt

@@ -26,7 +26,7 @@ interface CdcPartitionsCreatorDebeziumOperations<T : Comparable<T>> {
     fun generateColdStartOffset(): DebeziumOffset
 
     /** Generates Debezium properties for use with a [DebeziumColdStartingState]. */
-    fun generateColdStartProperties(): Map<String, String>
+    fun generateColdStartProperties(streams: List<Stream>): Map<String, String>
 
     /** Maps an incumbent [OpaqueStateValue] into a [DebeziumWarmStartState]. */
     fun deserializeState(opaqueStateValue: OpaqueStateValue): DebeziumWarmStartState

airbyte-cdk/bulk/toolkits/extract-cdc/src/test/kotlin/io/airbyte/cdk/read/cdc/CdcPartitionReaderMongoTest.kt

@@ -120,7 +120,7 @@ class CdcPartitionReaderMongoTest :
             return DebeziumOffset(mapOf(key to value))
         }
 
-        override fun generateColdStartProperties(): Map<String, String> =
+        override fun generateColdStartProperties(streams: List<Stream>): Map<String, String> =
             DebeziumPropertiesBuilder()
                 .withDefault()
                 .withConnector(MongoDbConnector::class.java)
@@ -141,7 +141,7 @@ class CdcPartitionReaderMongoTest :
                 .buildMap()
 
         override fun generateWarmStartProperties(streams: List<Stream>): Map<String, String> =
-            generateColdStartProperties()
+            generateColdStartProperties(streams)
 
         fun currentResumeToken(): BsonDocument =
             container.withMongoDatabase { mongoDatabase: MongoDatabase ->

airbyte-cdk/bulk/toolkits/extract-cdc/src/test/kotlin/io/airbyte/cdk/read/cdc/CdcPartitionReaderMySQLTest.kt

@@ -113,7 +113,7 @@ class CdcPartitionReaderMySQLTest :
             return DebeziumOffset(mapOf(key to value))
         }
 
-        override fun generateColdStartProperties(): Map<String, String> =
+        override fun generateColdStartProperties(streams: List<Stream>): Map<String, String> =
             DebeziumPropertiesBuilder()
                 .with(generateWarmStartProperties(emptyList()))
                 .with("snapshot.mode", "recovery")

airbyte-cdk/bulk/toolkits/extract-cdc/src/test/kotlin/io/airbyte/cdk/read/cdc/CdcPartitionReaderPostgresTest.kt

@@ -115,7 +115,7 @@ class CdcPartitionReaderPostgresTest :
                 .withStreams(streams)
                 .buildMap()
 
-        override fun generateColdStartProperties(): Map<String, String> =
+        override fun generateColdStartProperties(streams: List<Stream>): Map<String, String> =
             generateWarmStartProperties(emptyList())
 
         override fun generateColdStartOffset(): DebeziumOffset {

airbyte-cdk/bulk/toolkits/extract-cdc/src/test/kotlin/io/airbyte/cdk/read/cdc/CdcPartitionsCreatorTest.kt

@@ -50,8 +50,8 @@ class CdcPartitionsCreatorTest {
             configuredPrimaryKey = null,
             configuredCursor = TestMetaFieldDecorator.GlobalCursor,
         )
-
-    val global = Global(listOf(stream))
+    val streams = listOf(stream)
+    val global = Global(streams)
 
     val lowerBoundReference = AtomicReference<CreatorPosition>(null)
     val upperBoundReference = AtomicReference<CreatorPosition>(null)
@@ -80,8 +80,8 @@ class CdcPartitionsCreatorTest {
         every { creatorOps.position(syntheticOffset) } returns 123L
         every { creatorOps.position(incumbentOffset) } returns 123L
         every { creatorOps.generateColdStartOffset() } returns syntheticOffset
-        every { creatorOps.generateColdStartProperties() } returns emptyMap()
-        every { creatorOps.generateWarmStartProperties(listOf(stream)) } returns emptyMap()
+        every { creatorOps.generateColdStartProperties(streams) } returns emptyMap()
+        every { creatorOps.generateWarmStartProperties(streams) } returns emptyMap()
     }
 
     @Test

airbyte-cdk/bulk/toolkits/extract-cdc/src/testFixtures/kotlin/io/airbyte/cdk/read/cdc/AbstractCdcPartitionReaderTest.kt

@@ -83,7 +83,7 @@ abstract class AbstractCdcPartitionReaderTest<T : Comparable<T>, C : AutoCloseab
         container.createStream()
         val i0 =
             ReadInput(
-                debeziumOperations.generateColdStartProperties(),
+                debeziumOperations.generateColdStartProperties(listOf(stream)),
                 debeziumOperations.generateColdStartOffset(),
                 schemaHistory = null,
                 isSynthetic = true,

airbyte-integrations/connectors/source-mysql/metadata.yaml

@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
-  dockerImageTag: 3.11.6
+  dockerImageTag: 3.11.7
   dockerRepository: airbyte/source-mysql
   documentationUrl: https://docs.airbyte.com/integrations/sources/mysql
   githubIssueLabel: source-mysql

airbyte-integrations/connectors/source-mysql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDebeziumOperations.kt

@@ -342,7 +342,7 @@ class MySqlSourceDebeziumOperations(
         }
     }
 
-    override fun generateColdStartProperties(): Map<String, String> =
+    override fun generateColdStartProperties(streams: List<Stream>): Map<String, String> =
         DebeziumPropertiesBuilder()
             .with(commonProperties)
             // https://debezium.io/documentation/reference/2.2/connectors/mysql.html#mysql-property-snapshot-mode

docs/integrations/sources/mysql.md

@@ -226,6 +226,7 @@ Any database or table encoding combination of charset and collation is supported
 
 | Version     | Date       | Pull Request                                               | Subject                                                                                                                                         |
 |:------------|:-----------|:-----------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------|
+| 3.11.7 | 2025-03-12 | [55734](https://github.com/airbytehq/airbyte/pull/55734) | Expose additional stream context for debezium properties at startup |
 | 3.11.6 | 2025-03-06 | [55237](https://github.com/airbytehq/airbyte/pull/55237) | [Fix fetching binlog status for version >=8.4](https://github.com/airbytehq/airbyte/pull/55237#top) |
 | 3.11.5 | 2025-03-06 | [55234](https://github.com/airbytehq/airbyte/pull/55234) | Update base image version for certified DB source connectors |
 | 3.11.4 | 2025-03-06 | [55214](https://github.com/airbytehq/airbyte/pull/55214) | Update default encryption method to 'required'.                                                                                                 | |