bulk-cdk-toolkits-extract-cdc,source-mysql-v2: flesh out CDC

airbyte-cdk/bulk/toolkits/extract-cdc/build.gradle

@@ -3,6 +3,7 @@ dependencies {
     implementation project(':airbyte-cdk:bulk:core:bulk-cdk-core-extract')
     implementation 'io.debezium:debezium-api:2.7.1.Final'
     implementation 'io.debezium:debezium-embedded:2.7.1.Final'
+    implementation 'io.debezium:debezium-storage-file:2.7.1.Final'
 
     implementation 'commons-io:commons-io:2.17.0'
 }

airbyte-cdk/bulk/toolkits/extract-cdc/src/main/kotlin/io/airbyte/cdk/command/CdcSourceConfiguration.kt

@@ -4,5 +4,30 @@
 
 package io.airbyte.cdk.command
 
-/** Subtype of [SourceConfiguration] for CDC sources. */
-interface CdcSourceConfiguration : SourceConfiguration {}
+import io.micronaut.context.annotation.Factory
+import jakarta.inject.Singleton
+import java.time.Duration
+
+/** CDC-specific flavor of [SourceConfiguration]. */
+interface CdcSourceConfiguration : SourceConfiguration {
+
+    /**
+     * How often Debezium is to emit heartbeat events.
+     *
+     * This value is typically much lower in tests than in production.
+     */
+    val debeziumHeartbeatInterval: Duration
+
+    /**
+     * Micronaut factory which glues [ConfigurationJsonObjectSupplier] and
+     * [SourceConfigurationFactory] together to produce a [CdcSourceConfiguration] singleton.
+     */
+    @Factory
+    private class MicronautFactory {
+        @Singleton
+        fun <I : ConfigurationJsonObjectBase> cdcSourceConfig(
+            pojoSupplier: ConfigurationJsonObjectSupplier<I>,
+            factory: SourceConfigurationFactory<I, out CdcSourceConfiguration>,
+        ): CdcSourceConfiguration = factory.make(pojoSupplier.get())
+    }
+}