feat(standalone-kafka): explicitly use netty grpc server server, and …

…use loopback address for client and server (#690)
Trendyol · Jan 27, 2025 · dbfd9dd · dbfd9dd
1 parent 276fad7
commit dbfd9dd
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 4 deletions.
diff --git a/...-e2e-kafka/src/main/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/KafkaSystem.kt b/...-e2e-kafka/src/main/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/KafkaSystem.kt
@@ -8,7 +8,8 @@ import com.trendyol.stove.testing.e2e.standalone.kafka.intercepting.*
 import com.trendyol.stove.testing.e2e.system.TestSystem
 import com.trendyol.stove.testing.e2e.system.abstractions.*
 import com.trendyol.stove.testing.e2e.system.annotations.StoveDsl
-import io.grpc.*
+import io.grpc.Server
+import io.grpc.netty.NettyServerBuilder
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.selects.*
@@ -17,6 +18,7 @@ import org.apache.kafka.clients.consumer.*
 import org.apache.kafka.clients.producer.*
 import org.apache.kafka.common.serialization.*
 import org.slf4j.*
+import java.net.*
 import java.util.*
 import kotlin.reflect.KClass
 import kotlin.time.*
@@ -385,8 +387,8 @@ class KafkaSystem(
   private suspend fun startGrpcServer(): Server {
     System.setProperty(STOVE_KAFKA_BRIDGE_PORT, context.options.bridgeGrpcServerPort.toString())
     return Try {
-      ServerBuilder
-        .forPort(context.options.bridgeGrpcServerPort)
+      NettyServerBuilder
+        .forAddress(InetSocketAddress(InetAddress.getLoopbackAddress(), context.options.bridgeGrpcServerPort))
         .executor(StoveKafkaCoroutineScope.also { it.ensureActive() }.asExecutor)
         .addService(StoveKafkaObserverGrpcServer(sink))
         .handshakeTimeout(GRPC_TIMEOUT_IN_SECONDS, java.util.concurrent.TimeUnit.SECONDS)

diff --git a/...src/main/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/intercepting/GrpcUtils.kt b/...src/main/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/intercepting/GrpcUtils.kt
@@ -1,9 +1,12 @@
+@file:Suppress("HttpUrlsUsage")
+
 package com.trendyol.stove.testing.e2e.standalone.kafka.intercepting
 
 import com.squareup.wire.*
 import com.trendyol.stove.testing.e2e.standalone.kafka.*
 import kotlinx.coroutines.CoroutineScope
 import okhttp3.*
+import java.net.InetAddress
 import kotlin.time.Duration.Companion.seconds
 import kotlin.time.toJavaDuration
 
@@ -23,7 +26,9 @@ object GrpcUtils {
   fun createClient(onPort: String, scope: CoroutineScope): StoveKafkaObserverServiceClient = GrpcClient
     .Builder()
     .client(getClient(scope))
-    .baseUrl("http://0.0.0.0:$onPort".toHttpUrl())
+    .baseUrl(onLoopback(onPort))
     .build()
     .create<StoveKafkaObserverServiceClient>()
+
+  private fun onLoopback(port: String): GrpcHttpUrl = "http://${InetAddress.getLoopbackAddress().hostAddress}:$port".toHttpUrl()
 }