create a custom protocol for each network address for Kafka

martypitt · martypitt · commit edfcd0664622 · 2025-08-26T21:24:31.000+01:00
diff --git a/nebula-cli/src/main/kotlin/com/orbitalhq/nebula/cli/NebulaCli.kt b/nebula-cli/src/main/kotlin/com/orbitalhq/nebula/cli/NebulaCli.kt
@@ -1,5 +1,6 @@
 package com.orbitalhq.nebula.cli
 
+import com.orbitalhq.nebula.HostConfig
 import com.orbitalhq.nebula.NebulaConfig
 import com.orbitalhq.nebula.NebulaStackWithSource
 import com.orbitalhq.nebula.StackRunner
@@ -87,7 +88,7 @@ class Nebula : Callable<Int> {
 
         val scriptRunner = NebulaScriptExecutor()
         val stack = scriptRunner.runScript(file)
-        val stackWithSource = NebulaStackWithSource(stack, file.readText())
+        val stackWithSource = NebulaStackWithSource(stack, file.readText(), HostConfig.UNKNOWN)
 
         val stackRunner = StackRunner(nebulaConfig)
         stackRunner.submit(stackWithSource)
diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/InfrastructureComponent.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/InfrastructureComponent.kt
@@ -22,7 +22,7 @@ interface InfrastructureComponent<T> {
      * Used for display / diagnostics
      */
     val type: ComponentType
-    fun start(nebulaConfig: NebulaConfig): ComponentInfo<T>
+    fun start(nebulaConfig: NebulaConfig, hostConfig: HostConfig = HostConfig.UNKNOWN): ComponentInfo<T>
     fun stop()
 
     val componentInfo: ComponentInfo<T>?
diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/NebulaStack.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/NebulaStack.kt
@@ -18,7 +18,8 @@ typealias StackName = String
 
 data class NebulaStackWithSource(
     val stack: NebulaStack,
-    val source: String
+    val source: String,
+    val hostConfig: HostConfig
 ) {
     fun withName(id: String):NebulaStackWithSource {
         return this.copy(stack = stack.withName(id))
@@ -54,10 +55,10 @@ class NebulaStack(
         return component
     }
 
-    fun startComponents(config: NebulaConfig): Map<String, ComponentInfo<out Any?>> {
+    fun startComponents(config: NebulaConfig, hostConfig: HostConfig): Map<String, ComponentInfo<out Any?>> {
         stackStateEventSource.listenForEvents(name, components)
         return components.associate { component ->
-            component.type to component.start(config)
+            component.type to component.start(config, hostConfig)
         }
     }
 
diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/StackRunner.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/StackRunner.kt
@@ -13,6 +13,21 @@ data class NebulaConfig(
     val networkName: String = "nebula_network",
     val network: Network = Network.newNetwork()
 )
+
+/**
+ * Captures the details of the host, as captured when a stack
+ * is submitted.
+ * We can use this to determine things like the IP address that the
+ * stack is accessible via, to configure components who are strict
+ * about which network interfaces they listen on (eg., Kafka)
+ */
+data class HostConfig(
+    val hostAddresses:List<String>
+) {
+    companion object {
+        val UNKNOWN = HostConfig(emptyList())
+    }
+}
 class StackRunner(private val config: NebulaConfig = NebulaConfig()) {
     private val logger = KotlinLogging.logger {}
     val stacks = ConcurrentHashMap<StackName, NebulaStackWithSource>()
@@ -63,12 +78,14 @@ class StackRunner(private val config: NebulaConfig = NebulaConfig()) {
 
 
     private fun start(name: String) {
-        val stack = this.stacks[name]?.stack ?: error("Stack $name not found")
+        val stackWithSource = this.stacks[name] ?: error("Stack $name not found")
+        val stack = stackWithSource.stack
+        val hostConfig = stackWithSource.hostConfig
         stack.lifecycleEvents.subscribe { event ->
             logger.info { event.toString() }
         }
         logger.info { "Starting ${stack.name}" }
-        val state: Map<String, ComponentInfo<out Any?>> = stack.startComponents(config)
+        val state: Map<String, ComponentInfo<out Any?>> = stack.startComponents(config, hostConfig)
 
         _stackState[name] = state
     }
@@ -114,7 +131,7 @@ fun NebulaStackWithSource.start(): StackRunner {
  */
 fun NebulaStack.start(): StackRunner {
     val executor = StackRunner()
-    val stackWithSource = NebulaStackWithSource(this, "Source not provided - Random UUID follows - ${UUID.randomUUID()}" )
+    val stackWithSource = NebulaStackWithSource(this, "Source not provided - Random UUID follows - ${UUID.randomUUID()}" , hostConfig = HostConfig.UNKNOWN)
     executor.submit(stackWithSource)
     return executor
 }
diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/hazelcast/HazelcastExecutor.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/hazelcast/HazelcastExecutor.kt
@@ -1,5 +1,6 @@
 package com.orbitalhq.nebula.hazelcast
 
+import com.orbitalhq.nebula.HostConfig
 import com.orbitalhq.nebula.InfrastructureComponent
 import com.orbitalhq.nebula.NebulaConfig
 import com.orbitalhq.nebula.StackRunner
@@ -34,7 +35,7 @@ class HazelcastExecutor(private val config: HazelcastConfig) : InfrastructureCom
     override val type: ComponentType = "hazelcast"
     private val eventSource = ComponentLifecycleEventSource()
 
-    override fun start(nebulaConfig: NebulaConfig): ComponentInfo<HazelcastContainerConfig> {
+    override fun start(nebulaConfig: NebulaConfig, hostConfig: HostConfig): ComponentInfo<HazelcastContainerConfig> {
         eventSource.starting()
         container = GenericContainer(DockerImageName.parse(config.imageName))
             .withExposedPorts(5701)
diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/http/HttpExecutor.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/http/HttpExecutor.kt
@@ -1,5 +1,6 @@
 package com.orbitalhq.nebula.http
 
+import com.orbitalhq.nebula.HostConfig
 import com.orbitalhq.nebula.InfrastructureComponent
 import com.orbitalhq.nebula.NebulaConfig
 import com.orbitalhq.nebula.StackRunner
@@ -47,7 +48,7 @@ class HttpExecutor(private val config: HttpConfig) : InfrastructureComponent<Htt
     override var componentInfo: ComponentInfo<HttpServerConfig>? = null
         private set
 
-    override fun start(nebulaConfig: NebulaConfig):ComponentInfo<HttpServerConfig> {
+    override fun start(nebulaConfig: NebulaConfig, hostConfig: HostConfig):ComponentInfo<HttpServerConfig> {
         eventSource.starting()
         server = embeddedServer(Netty, port = port) {
             routing {
diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/kafka/KafkaExecutor.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/kafka/KafkaExecutor.kt
@@ -1,5 +1,6 @@
 package com.orbitalhq.nebula.kafka
 
+import com.orbitalhq.nebula.HostConfig
 import com.orbitalhq.nebula.InfrastructureComponent
 import com.orbitalhq.nebula.NebulaConfig
 import com.orbitalhq.nebula.StackRunner
@@ -10,6 +11,7 @@ import com.orbitalhq.nebula.core.HostNameAwareContainerConfig
 import com.orbitalhq.nebula.events.ComponentLifecycleEventSource
 import com.orbitalhq.nebula.utils.updateHostReferences
 import io.github.oshai.kotlinlogging.KotlinLogging
+import io.ktor.http.*
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -58,10 +60,15 @@ class KafkaExecutor(private val config: KafkaConfig) : InfrastructureComponent<K
     override var componentInfo: ComponentInfo<KafkaContainerConfig>? = null
         private set
 
-    override fun start(nebulaConfig: NebulaConfig): ComponentInfo<KafkaContainerConfig> {
+    override fun start(nebulaConfig: NebulaConfig, hostConfig: HostConfig): ComponentInfo<KafkaContainerConfig> {
+
         kafkaContainer = KafkaContainer(DockerImageName.parse(config.imageName))
+            .let { container ->
+                configureExternalListenerAddresses(hostConfig, container)
+            }
             .withNetwork(nebulaConfig.network)
             .withNetworkAliases(config.componentName)
+
         eventSource.startContainerAndEmitEvents(kafkaContainer)
 
         val bootstrapServers = kafkaContainer.bootstrapServers
@@ -104,7 +111,8 @@ class KafkaExecutor(private val config: KafkaConfig) : InfrastructureComponent<K
         componentInfo = ComponentInfo(
             containerInfoFrom(kafkaContainer),
             KafkaContainerConfig(
-                kafkaContainer.bootstrapServers
+                kafkaContainer.bootstrapServers,
+                addressToProtocol
             ),
             type = type,
             name = name,
@@ -115,6 +123,41 @@ class KafkaExecutor(private val config: KafkaConfig) : InfrastructureComponent<K
         eventSource.running()
         return componentInfo!!
     }
+    private val addressToProtocol = mutableMapOf<String,String>()
+
+    /**
+     * Configures listeners for the external addresses that
+     * we're known by. Otherwise consumers who are connecting from
+     * an external IP address can't connect to Kafka.
+     *
+     * Works by establishing a custom protocol mapping (test containers adds a TC-0 / TC-1 etc.. to the front)
+     */
+    private fun configureExternalListenerAddresses(
+        hostConfig: HostConfig,
+        container: KafkaContainer
+    ): KafkaContainer {
+        hostConfig.hostAddresses.forEachIndexed { index, externalAddress ->
+            // each custom address gets a port incrementally higher than 9094
+            val port = 9094 + index
+
+            container.addExposedPort(port)
+            container.withListener {
+                try {
+                    // If this doesn't throw an error, we're building the KAFKA_ADVERTISED_LISTENER part
+                    val mappedPort = container.getMappedPort(port)
+                    logger.info { "Configuring external Kafka listener for $externalAddress:$port" }
+                    // Test containers maps these as TC-0://0.0.0.0:909X
+                    val protocol = "TC-$index"
+                    addressToProtocol.put(externalAddress, "$protocol://$externalAddress:$mappedPort")
+                    "$externalAddress:$mappedPort"
+                } catch (e: IllegalStateException) {
+                    // The container isn't ready yet, so we're building the internal KAFKA_LISTENER
+                    "0.0.0.0:$port"
+                }
+            }
+        }
+        return container
+    }
 
     override fun stop() {
         eventSource.stopping()
@@ -146,15 +189,15 @@ class KafkaExecutor(private val config: KafkaConfig) : InfrastructureComponent<K
         val adminProps = Properties().apply {
             put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
         }
-        
+
         AdminClient.create(adminProps).use { adminClient ->
             // Group topics by name to handle duplicates and use max partitions
             val topicConfigs = producers.groupBy { it.topic }
                 .mapValues { (_, configs) -> configs.maxOf { it.partitions } }
                 .map { (topicName, partitions) ->
                     NewTopic(topicName, partitions, 1.toShort()) // replication factor of 1 for single broker
                 }
-            
+
             if (topicConfigs.isNotEmpty()) {
                 try {
                     val result = adminClient.createTopics(topicConfigs)
@@ -170,8 +213,14 @@ class KafkaExecutor(private val config: KafkaConfig) : InfrastructureComponent<K
     }
 }
 
-data class KafkaContainerConfig(val bootstrapServers: String) : HostNameAwareContainerConfig<KafkaContainerConfig> {
+data class KafkaContainerConfig(val bootstrapServers: String, private val externalAddressToListenerProtocol:Map<String,String>) : HostNameAwareContainerConfig<KafkaContainerConfig> {
     override fun updateHostReferences(containerHost: String, publicHost: String): KafkaContainerConfig {
-        return copy(bootstrapServers = bootstrapServers.updateHostReferences(containerHost, publicHost))
+        // If we configured a dedicated bootstrap server protocol for this host, then use that
+        val dedicatedBootstrapServer = externalAddressToListenerProtocol[publicHost]
+        return if (dedicatedBootstrapServer != null) {
+            copy(bootstrapServers = dedicatedBootstrapServer)
+        } else {
+            this
+        }
     }
 }
diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/mongo/MongoExecutor.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/mongo/MongoExecutor.kt
@@ -1,6 +1,7 @@
 package com.orbitalhq.nebula.mongo
 
 import com.mongodb.client.MongoClients
+import com.orbitalhq.nebula.HostConfig
 import com.orbitalhq.nebula.InfrastructureComponent
 import com.orbitalhq.nebula.NebulaConfig
 import com.orbitalhq.nebula.StackRunner
@@ -34,7 +35,7 @@ class MongoExecutor(private val config: MongoConfig) : InfrastructureComponent<M
         private val logger = KotlinLogging.logger {}
     }
 
-    override fun start(nebulaConfig: NebulaConfig): ComponentInfo<MongoContainerConfig> {
+    override fun start(nebulaConfig: NebulaConfig, hostConfig: HostConfig): ComponentInfo<MongoContainerConfig> {
         mongoContainer = MongoDBContainer(DockerImageName.parse(config.imageName))
             .withNetwork(nebulaConfig.network)
             .withNetworkAliases(config.componentName)
diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/s3/S3Executor.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/s3/S3Executor.kt
@@ -1,5 +1,6 @@
 package com.orbitalhq.nebula.s3
 
+import com.orbitalhq.nebula.HostConfig
 import com.orbitalhq.nebula.InfrastructureComponent
 import com.orbitalhq.nebula.NebulaConfig
 import com.orbitalhq.nebula.StackRunner
@@ -43,7 +44,7 @@ class S3Executor(private val config: S3Config) : InfrastructureComponent<Localst
     override var componentInfo: ComponentInfo<LocalstackContainerConfig>? = null
         private set
 
-    override fun start(nebulaConfig: NebulaConfig):ComponentInfo<LocalstackContainerConfig> {
+    override fun start(nebulaConfig: NebulaConfig, hostConfig: HostConfig):ComponentInfo<LocalstackContainerConfig> {
         localstack = LocalStackContainer(DockerImageName.parse(config.imageName))
             .withServices(LocalStackContainer.Service.S3)
             .withNetwork(nebulaConfig.network)
diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/sql/DatabaseExecutor.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/sql/DatabaseExecutor.kt
@@ -1,5 +1,6 @@
 package com.orbitalhq.nebula.sql
 
+import com.orbitalhq.nebula.HostConfig
 import com.orbitalhq.nebula.InfrastructureComponent
 import com.orbitalhq.nebula.NebulaConfig
 import com.orbitalhq.nebula.StackRunner
@@ -45,7 +46,7 @@ class DatabaseExecutor(private val config: DatabaseConfig) : InfrastructureCompo
     override var componentInfo: ComponentInfo<DatabaseContainerConfig>? = null
         private set
 
-    override fun start(nebulaConfig: NebulaConfig): ComponentInfo<DatabaseContainerConfig> {
+    override fun start(nebulaConfig: NebulaConfig, hostConfig: HostConfig): ComponentInfo<DatabaseContainerConfig> {
         databaseContainer = config.container.withDatabaseName(config.databaseName)
             .withNetwork(nebulaConfig.network)
             .withNetworkAliases(config.componentName)
diff --git a/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/taxi/TaxiPublishingExecutor.kt b/nebula-dsl/src/main/kotlin/com/orbitalhq/nebula/taxi/TaxiPublishingExecutor.kt
@@ -1,5 +1,6 @@
 package com.orbitalhq.nebula.taxi
 
+import com.orbitalhq.nebula.HostConfig
 import com.orbitalhq.nebula.InfrastructureComponent
 import com.orbitalhq.nebula.NebulaConfig
 import com.orbitalhq.nebula.core.ComponentInfo
@@ -29,7 +30,7 @@ class TaxiPublishingExecutor(private val config: TaxiPublisherConfig) : Infrastr
         private set
 
 
-    override fun start(nebulaConfig: NebulaConfig): ComponentInfo<TaxiPublisherConfig> {
+    override fun start(nebulaConfig: NebulaConfig, hostConfig: HostConfig): ComponentInfo<TaxiPublisherConfig> {
         eventSource.starting()
 
         // Publish
diff --git a/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/NebulaScriptExecutor.kt b/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/NebulaScriptExecutor.kt
@@ -1,5 +1,6 @@
 package com.orbitalhq.nebula.runtime
 
+import com.orbitalhq.nebula.HostConfig
 import com.orbitalhq.nebula.NebulaScript
 import com.orbitalhq.nebula.NebulaStack
 import com.orbitalhq.nebula.NebulaStackWithSource
@@ -56,9 +57,9 @@ class NebulaScriptExecutor {
         val source = string.toScriptSource()
         return runScript(source)
     }
-    fun toStackWithSource(string: String): NebulaStackWithSource {
+    fun toStackWithSource(string: String, hostConfig: HostConfig): NebulaStackWithSource {
         val source = string.toScriptSource()
         val stack = runScript(source)
-        return NebulaStackWithSource(stack, string)
+        return NebulaStackWithSource(stack, string, hostConfig)
     }
 }
diff --git a/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/server/NebulaServer.kt b/nebula-runtime/src/main/kotlin/com/orbitalhq/nebula/runtime/server/NebulaServer.kt
@@ -3,6 +3,7 @@ package com.orbitalhq.nebula.runtime.server
 import com.fasterxml.jackson.databind.SerializationFeature
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import com.fasterxml.jackson.module.kotlin.readValue
+import com.orbitalhq.nebula.HostConfig
 import com.orbitalhq.nebula.NebulaConfig
 import com.orbitalhq.nebula.NebulaStack
 import com.orbitalhq.nebula.NebulaStackWithSource
@@ -67,7 +68,7 @@ class NebulaServer(
                     // Create a stack without an id -- an id is assigned
                     post {
                         val script = call.receiveText()
-                        val stack = scriptExecutor.toStackWithSource(script)
+                        val stack = scriptExecutor.toStackWithSource(script, call.hostConfig())
                         stackExecutor.submit(stack, startAsync = true)
                         call.respond(stack.name)
                     }
@@ -77,7 +78,7 @@ class NebulaServer(
                             "Missing or malformed id"
                         )
                         val script = call.receiveText()
-                        val stack = scriptExecutor.toStackWithSource(script).let { stack ->
+                        val stack = scriptExecutor.toStackWithSource(script, call.hostConfig()).let { stack ->
                             stack.withName(id)
                         }
                         stackExecutor.submit(stack)
@@ -111,7 +112,7 @@ class NebulaServer(
                         val updateStacksRequest =
                             objectMapper.readValue<UpdateStackRSocketRequest>(payloadJson)
 
-                        val stackMap = compile(updateStacksRequest)
+                        val stackMap = compile(updateStacksRequest, call.hostConfig())
                         val eventStreams = stackMap.map { (name, stack) ->
                             stackExecutor.submit(stack, name, startAsync = true)
                         }
@@ -131,9 +132,9 @@ class NebulaServer(
         }.start(wait = wait)
     }
 
-    private fun compile(updateStacksRequest: UpdateStackRSocketRequest): Map<StackName, NebulaStackWithSource> {
+    private fun compile(updateStacksRequest: UpdateStackRSocketRequest, hostConfig: HostConfig): Map<StackName, NebulaStackWithSource> {
         return updateStacksRequest.stacks.mapValues { (key, stackScript) ->
-            scriptExecutor.toStackWithSource(stackScript).withName(key)
+            scriptExecutor.toStackWithSource(stackScript, hostConfig).withName(key)
         }
     }
 
@@ -143,4 +144,8 @@ data class StackEventStreamRequest(val stackId: StackName)
 
 typealias StackScript = String
 
-data class UpdateStackRSocketRequest(val stacks: Map<StackName, StackScript>)
+data class UpdateStackRSocketRequest(val stacks: Map<StackName, StackScript>)
+
+fun ApplicationCall.hostConfig():HostConfig {
+    return HostConfig(listOf(this.request.host()))
+}

Original file line number	Diff line number	Diff line change
`@@ -18,7 +18,8 @@ typealias StackName = String`
`18`	`18`
`19`	`19`	`data class NebulaStackWithSource(`
`20`	`20`	`val stack: NebulaStack,`
`21`		`- val source: String`
	`21`	`+ val source: String,`
	`22`	`+ val hostConfig: HostConfig`
`22`	`23`	`) {`
`23`	`24`	`fun withName(id: String):NebulaStackWithSource {`
`24`	`25`	`return this.copy(stack = stack.withName(id))`
`@@ -54,10 +55,10 @@ class NebulaStack(`
`54`	`55`	`return component`
`55`	`56`	`}`
`56`	`57`
`57`		`- fun startComponents(config: NebulaConfig): Map<String, ComponentInfo<out Any?>> {`
	`58`	`+ fun startComponents(config: NebulaConfig, hostConfig: HostConfig): Map<String, ComponentInfo<out Any?>> {`
`58`	`59`	`stackStateEventSource.listenForEvents(name, components)`
`59`	`60`	`return components.associate { component ->`
`60`		`- component.type to component.start(config)`
	`61`	`+ component.type to component.start(config, hostConfig)`
`61`	`62`	`}`
`62`	`63`	`}`
`63`	`64`