add default timeout for grpc call (#160)

Author: daleiz

client/src/main/java/io/hstream/impl/DefaultSettings.java

@@ -7,4 +7,6 @@ public class DefaultSettings {
   public static final int APPEND_RETRY_MAX_TIMES = 5;
 
   public static final String DEFAULT_PARTITION_KEY = "";
+
+  public static final long GRPC_CALL_TIMEOUT_SECONDS = 5;
 }

client/src/main/kotlin/io/hstream/impl/HStreamClientKtImpl.kt

@@ -35,6 +35,7 @@ import io.hstream.util.GrpcUtils
 import kotlinx.coroutines.runBlocking
 import org.slf4j.LoggerFactory
 import java.util.concurrent.CompletableFuture
+import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicReference
 import kotlin.streams.toList
 
@@ -62,6 +63,10 @@ class HStreamClientKtImpl(bootstrapServerUrls: List<String>, credentials: Channe
         return HStreamApiGrpcKt.HStreamApiCoroutineStub(channelProvider.get(url))
     }
 
+    fun getCoroutineStubWithTimeout(url: String, timeoutSeconds: Long): HStreamApiGrpcKt.HStreamApiCoroutineStub {
+        return HStreamApiGrpcKt.HStreamApiCoroutineStub(channelProvider.get(url)).withDeadlineAfter(timeoutSeconds, TimeUnit.SECONDS)
+    }
+
     init {
         logger.info("client init with bootstrapServerUrls [{}]", bootstrapServerUrls)
         val describeClusterResponse = unaryCallWithCurrentUrls(

client/src/main/kotlin/io/hstream/impl/ProducerKtImpl.kt

@@ -137,7 +137,7 @@ open class ProducerKtImpl(private val client: HStreamClientKtImpl, private val s
         val serverUrl = lookupServerUrl(shardId, forceUpdate)
         logger.debug("try append with serverUrl [{}], current left tryTimes is [{}]", serverUrl, tryTimes)
         return try {
-            client.getCoroutineStub(serverUrl)
+            client.getCoroutineStubWithTimeout(serverUrl, DefaultSettings.GRPC_CALL_TIMEOUT_SECONDS)
                 .append(appendRequest).recordIdsList.map(GrpcUtils::recordIdFromGrpc)
         } catch (e: StatusException) {
             handleGRPCException(serverUrl, e)

client/src/main/kotlin/io/hstream/impl/ReaderKtImpl.kt

@@ -50,7 +50,7 @@ class ReaderKtImpl(
         readerScope.launch {
             try {
                 val readShardRequest = ReadShardRequest.newBuilder().setReaderId(readerId).setMaxRecords(maxRecords).build()
-                val readShardResponse = client.getCoroutineStub(serverUrl)
+                val readShardResponse = client.getCoroutineStubWithTimeout(serverUrl, DefaultSettings.GRPC_CALL_TIMEOUT_SECONDS)
                     .readShard(readShardRequest)
                 val res = readShardResponse.receivedRecordsList.flatMap {
                     RecordUtils.decompress(it).map { receivedHStreamRecord ->
@@ -82,7 +82,7 @@ class ReaderKtImpl(
             .setReaderId(readerId)
             .build()
         runBlocking {
-            client.getCoroutineStub(serverUrl).deleteShardReader(deleteShardReaderRequest)
+            client.getCoroutineStubWithTimeout(serverUrl, DefaultSettings.GRPC_CALL_TIMEOUT_SECONDS).deleteShardReader(deleteShardReaderRequest)
         }
 
         logger.info("Reader [{}] closed", readerId)

client/src/main/kotlin/io/hstream/impl/Utils.kt

@@ -17,6 +17,7 @@ import kotlinx.coroutines.future.future
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import java.util.concurrent.CompletableFuture
+import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.CoroutineContext
 
@@ -94,7 +95,7 @@ suspend fun <Resp> unaryCallCoroutine(urlsRef: AtomicReference<List<String>>, ch
     logger.debug("unary rpc with urls [{}]", urls)
 
     try {
-        return call(HStreamApiCoroutineStub(channelProvider.get(urls[0])))
+        return call(HStreamApiCoroutineStub(channelProvider.get(urls[0])).withDeadlineAfter(DefaultSettings.GRPC_CALL_TIMEOUT_SECONDS, TimeUnit.SECONDS))
     } catch (e: StatusException) {
         return handleGRPCException(urls, e)
     } catch (e: StatusRuntimeException) {