chore: add the speed mode selection to the replication input (#16425)

Co-authored-by: Jimmy Ma <[email protected]>

Author: benmoriceau

airbyte-server/src/main/kotlin/io/airbyte/server/services/JobInputService.kt

@@ -208,6 +208,8 @@ class JobInputService(
       heartbeatMaxSecondsBetweenMessages = sourceDefinition.maxSecondsBetweenMessages,
       supportsRefreshes = sourceDefinitionVersion.supportsRefreshes,
       schemaRefreshOutput = appliedCatalogDiff?.let { RefreshSchemaActivityOutput(appliedDiff = it) },
+      sourceIPCOptions = sourceDefinitionVersion.connectorIPCOptions,
+      destinationIPCOptions = destinationDefinitionVersion.connectorIPCOptions,
     )
   }
 

airbyte-server/src/test/kotlin/io/airbyte/server/services/JobInputServiceTest.kt

@@ -547,19 +547,32 @@ class JobInputServiceTest {
     every { mockSourceDefinition.custom } returns false
     every { mockSourceDefinition.sourceType } returns StandardSourceDefinition.SourceType.DATABASE
 
+    val sourceICPOption =
+      Jsons.jsonNode(
+        mapOf(
+          "source" to "ICPOption",
+        ),
+      )
     val mockSourceDefinitionVersion = mockk<ActorDefinitionVersion>()
     every { mockSourceDefinitionVersion.dockerRepository } returns dockerRepository
     every { mockSourceDefinitionVersion.dockerImageTag } returns dockerImageTag
     every { mockSourceDefinitionVersion.protocolVersion } returns protocolVersion
     every { mockSourceDefinitionVersion.allowedHosts } returns AllowedHosts()
     every { mockSourceDefinitionVersion.supportsRefreshes } returns false
+    every { mockSourceDefinitionVersion.connectorIPCOptions } returns sourceICPOption
 
     val mockDestination = mockk<DestinationConnection>()
     every { mockDestination.destinationId } returns destinationId
     every { mockDestination.destinationDefinitionId } returns destinationDefinitionId
     every { mockDestination.workspaceId } returns workspaceId
     every { mockDestination.configuration } returns destinationConfiguration
 
+    val destinationICPOption =
+      Jsons.jsonNode(
+        mapOf(
+          "destination" to "ICPOption",
+        ),
+      )
     val mockDestinationDefinition = mockk<StandardDestinationDefinition>()
     every { mockDestinationDefinition.destinationDefinitionId } returns destinationDefinitionId
     every { mockDestinationDefinition.custom } returns true
@@ -569,6 +582,7 @@ class JobInputServiceTest {
     every { mockDestinationDefinitionVersion.dockerRepository } returns destinationImage
     every { mockDestinationDefinitionVersion.protocolVersion } returns protocolVersion
     every { mockDestinationDefinitionVersion.allowedHosts } returns AllowedHosts()
+    every { mockDestinationDefinitionVersion.connectorIPCOptions } returns destinationICPOption
 
     val attemptSyncConfig = mockk<AttemptSyncConfig>()
     every { attemptSyncConfig.sourceConfiguration } returns null
@@ -667,6 +681,8 @@ class JobInputServiceTest {
         heartbeatMaxSecondsBetweenMessages = 3600L,
         supportsRefreshes = false,
         schemaRefreshOutput = RefreshSchemaActivityOutput(appliedCatalogDiff),
+        sourceIPCOptions = sourceICPOption,
+        destinationIPCOptions = destinationICPOption,
       )
 
     val actual = jobInputService.getReplicationInput(connectionId, appliedCatalogDiff, signalInput, jobId, attemptNumber.toLong())