I made a simple gRPC server in Kotlin with coroutines and a client with Java. In the cliente I enabled and configured a retry policy, but it does was not work. I speend a lot of time to find a solution, belivied that my client was broken, but the problem it was in the server. I will show you the code.
This is my proto file:
syntax = "proto3";
option java_multiple_files = true;
option java_package = "br.com.will.protoclasses";
option java_outer_classname = "NotificationProto";
package notification;
service Notification {
rpc SendPush (SendPushNotificationRequest) returns (SendPushNotificationResponse);
}
message SendPushNotificationRequest {
string title = 1;
string message = 2;
string customer_id = 3;
}
message SendPushNotificationResponse {
string message = 1;
}
This is the client:
open class NotificationClient(private val channel: ManagedChannel) {
private val stub: NotificationGrpcKt.NotificationCoroutineStub =
NotificationGrpcKt.NotificationCoroutineStub(channel)
suspend fun send() {
val request =
SendPushNotificationRequest.newBuilder().setCustomerId(UUID.randomUUID().toString()).setMessage("test")
.setTitle("test").build()
val response = stub.sendPush(request)
println("Received: ${response.message}")
}
}
suspend fun main(args: Array<String>) {
val port = System.getenv("PORT")?.toInt() ?: 50051
val retryPolicy: MutableMap<String, Any> = HashMap()
retryPolicy["maxAttempts"] = 5.0
retryPolicy["initialBackoff"] = "10s"
retryPolicy["maxBackoff"] = "30s"
retryPolicy["backoffMultiplier"] = 2.0
retryPolicy["retryableStatusCodes"] = listOf<Any>("INTERNAL")
val methodConfig: MutableMap<String, Any> = HashMap()
val name: MutableMap<String, Any> = HashMap()
name["service"] = "notification.Notification"
name["method"] = "SendPush"
methodConfig["name"] = listOf<Any>(name)
methodConfig["retryPolicy"] = retryPolicy
val serviceConfig: MutableMap<String, Any> = HashMap()
serviceConfig["methodConfig"] = listOf<Any>(methodConfig)
print(serviceConfig)
val channel = ManagedChannelBuilder.forAddress("localhost", port)
.usePlaintext()
.defaultServiceConfig(serviceConfig)
.enableRetry()
.build()
val client = NotificationClient(channel)
client.send()
}
This is a part of my gRPC service, where i was testing the retry policy (the retry policy on client does not work with this implementation):
override suspend fun sendPush(request: SendPushNotificationRequest): SendPushNotificationResponse {
val count: Int = retryCounter.incrementAndGet()
log.info("Received a call on method sendPushNotification with payload -> $request")
if (random.nextFloat() < UNAVAILABLE_PERCENTAGE) {
log.info("Returning stubbed INTERNAL error. count: $count")
throw Status.INTERNAL.withDescription("error").asRuntimeException()
}
log.info("Returning successful Hello response, count: $count")
return SendPushNotificationResponse.newBuilder().setMessage("success").build()
}
Another implementation, but now using StreamObserver (This implementation works fine):
override fun sendPush(
request: SendPushNotificationRequest?,
responseObserver: StreamObserver<SendPushNotificationResponse>?
) {
log.info("Received a call on method sendPushNotification with payload -> $request")
val count: Int = retryCounter.incrementAndGet()
if (random.nextFloat() < UNAVAILABLE_PERCENTAGE) {
log.info("Returning stubbed UNAVAILABLE error. count: $count")
responseObserver!!.onError(
Status.UNAVAILABLE.withDescription("error").asRuntimeException()
)
} else {
log.info("Returning successful Hello response, count: $count")
responseObserver!!.onNext(SendPushNotificationResponse.newBuilder().setMessage("success").build())
return responseObserver.onCompleted()
}
}
The question is, whats is wrong? Can someone help me?