[TH2-5178] implemented gRPC backpressure for the Execute method (#21)

* Updated libraries

Author: Nikita-Smirnov-Exactpro

.github/workflows/build-dev-release.yml

@@ -165,5 +165,4 @@ jobs:
     needs: [ app-version, publish-docker, build-and-publish-core-java ]
     uses: th2-net/.github/.github/workflows/compaund-git-tag-push.yml@main
     with:
-      runsOn: ${{ inputs.runsOn }}
       tagName: ${{ needs.app-version.outputs.version }}-dev

.github/workflows/build-release.yml

@@ -124,5 +124,4 @@ jobs:
     needs: [ app-version, publish-docker, build-and-publish-core-java ]
     uses: th2-net/.github/.github/workflows/compaund-git-tag-push.yml@main
     with:
-      runsOn: ${{ inputs.runsOn }}
       tagName: ${{ needs.app-version.outputs.version }}

README.md

@@ -1,4 +1,4 @@
-# th2-read-db 0.8.0
+# th2-read-db 0.9.0
 
 The read-db is a component for extracting data from databases using JDBC technology. If database has JDBC driver the read can work with the database
 
@@ -339,10 +339,25 @@ spec:
 
 ## Changes
 
+### 0.9.0
+
++ implemented gRPC backpressure for the `Execute` method
++ updated jdbc:
+  + mysql-connector-j:`8.3.0`
+  + ojdbc11:`23.3.0.23.09`
+  + postgresql:`42.7.3`
++ updated: 
+  + common:`5.10.0-dev`
+  + grpc-common:`4.4.0-dev`
+  + common-utils:`2.2.2-dev`
+  + common-utils:`2.2.2-dev`
+
 ### 0.8.0
 
 + implemented the `Load` gRPC method.
 + fixed the catching java Error such as OutOfMemoryError problem
++ updated bom:`4.6.0`
+  
 
 ### 0.7.0
 

app/build.gradle.kts

@@ -12,13 +12,13 @@ dependencies {
     implementation(project(":read-db-core"))
 
     //region Drivers
-    runtimeOnly("org.postgresql:postgresql:42.6.0") {
+    runtimeOnly("org.postgresql:postgresql:42.7.3") {
         because("prostresql support")
     }
-    runtimeOnly("com.mysql:mysql-connector-j:8.1.0") {
+    runtimeOnly("com.mysql:mysql-connector-j:8.3.0") {
         because("mysql support")
     }
-    runtimeOnly("com.oracle.database.jdbc:ojdbc11:23.2.0.0") {
+    runtimeOnly("com.oracle.database.jdbc:ojdbc11:23.3.0.23.09") {
         because("oracle support")
     }
     runtimeOnly("com.microsoft.sqlserver:mssql-jdbc:12.4.0.jre11") {

app/gradle.properties

@@ -1,3 +1,3 @@
 kotlin.code.style=official
-release_version=0.8.0
+release_version=0.9.0
 description=read-db component for extracting data from databases using JDBC technology

core/build.gradle.kts

@@ -18,8 +18,8 @@ configurations.all {
 dependencies {
     implementation(project(":grpc-read-db"))
 
-    implementation("com.exactpro.th2:common:5.7.1-dev")
-    implementation("com.exactpro.th2:common-utils:2.2.0-dev")
+    implementation("com.exactpro.th2:common:5.10.0-dev")
+    implementation("com.exactpro.th2:common-utils:2.2.2-dev")
     implementation("com.exactpro.th2:lw-data-provider-utils:0.0.1-dev")
 
     implementation("org.slf4j:slf4j-api")
@@ -28,7 +28,7 @@ dependencies {
         because("connection pool")
     }
     implementation("org.apache.commons:commons-text")
-    implementation("com.opencsv:opencsv:5.8") {
+    implementation("com.opencsv:opencsv:5.9") {
         because("publishes raw messages in csv format")
     }
 
@@ -42,7 +42,7 @@ dependencies {
     testImplementation("org.mockito.kotlin:mockito-kotlin:5.1.0")
     testImplementation("io.strikt:strikt-core:0.34.1")
 
-    testImplementation(platform("org.testcontainers:testcontainers-bom:1.19.0"))
+    testImplementation(platform("org.testcontainers:testcontainers-bom:1.19.7"))
     testImplementation("org.testcontainers:testcontainers")
     testImplementation("org.testcontainers:mysql")
     testImplementation("org.testcontainers:oracle-xe")
@@ -51,10 +51,10 @@ dependencies {
 
     testImplementation("com.exactpro.th2:junit-jupiter-integration:0.0.1")
 
-    testRuntimeOnly("com.mysql:mysql-connector-j:8.1.0") {
+    testRuntimeOnly("com.mysql:mysql-connector-j:8.3.0") {
         because("mysql support")
     }
-    testRuntimeOnly("com.oracle.database.jdbc:ojdbc11:23.2.0.0") {
+    testRuntimeOnly("com.oracle.database.jdbc:ojdbc11:23.3.0.23.09") {
         because("oracle support")
     }
 }

core/gradle.properties

@@ -1,3 +1,3 @@
 kotlin.code.style=official
-release_version=0.8.0
+release_version=0.9.0
 description=core part of read db to create an application with required JDBC drivers in the classpath

core/src/main/kotlin/com/exactpro/th2/read/db/app/DataBaseReader.kt

@@ -157,7 +157,7 @@ class DataBaseReader(
         LOGGER.info { "Reader closed" }
     }
 
-    private fun TableRow.transferTo(sourceId: DataSourceId, vararg listeners: RowListener) {
+    private suspend fun TableRow.transferTo(sourceId: DataSourceId, vararg listeners: RowListener) {
         listeners.forEach { listener ->
             runCatchingException { listener.onRow(sourceId, this) }.onFailure {
                 LOGGER.error(it) { "error during row processing by listener ${listener::class}" }

core/src/main/kotlin/com/exactpro/th2/read/db/core/ResultListener.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 Exactpro (Exactpro Systems Limited)
+ * Copyright 2022-2024 Exactpro (Exactpro Systems Limited)
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,5 +22,5 @@ interface ResultListener : RowListener {
 }
 
 fun interface RowListener {
-    fun onRow(sourceId: DataSourceId, row: TableRow)
+    suspend fun onRow(sourceId: DataSourceId, row: TableRow)
 }

core/src/main/kotlin/com/exactpro/th2/read/db/impl/grpc/DataBaseReaderGrpcServer.kt

@@ -44,7 +44,11 @@ import com.exactpro.th2.read.db.impl.grpc.util.toModel
 import com.google.protobuf.Empty
 import com.google.protobuf.Message
 import io.grpc.Status
+import io.grpc.StatusRuntimeException
+import io.grpc.stub.ServerCallStreamObserver
 import io.grpc.stub.StreamObserver
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ClosedReceiveChannelException
 import mu.KotlinLogging
 import java.time.Instant
 import java.util.concurrent.TimeUnit
@@ -58,8 +62,8 @@ class DataBaseReaderGrpcServer(
     private val onEvent: OnEvent,
 ) : ReadDbGrpc.ReadDbImplBase() {
     override fun execute(request: QueryRequest, responseObserver: StreamObserver<QueryResponse>) {
-        execute("execute", request, responseObserver) { event, parentEventId, _ ->
-            GrpcExecuteListener(responseObserver, event) {
+        execute("execute", request, responseObserver) { event, parentEventId, executionId ->
+            GrpcExecuteListener(responseObserver, executionId, event) {
                 onEvent.accept(it, parentEventId)
             }
         }
@@ -155,8 +159,8 @@ class DataBaseReaderGrpcServer(
             event.bodyData(executeQueryRequest.toBody(executionId))
             app.executeQuery(
                 executeQueryRequest,
-                createListener(event, parentEventId, executionId),
-                ) { row ->
+                createListener(event, parentEventId, executionId)
+            ) { row ->
                 row.copy(associatedMessageType = associatedMessageType, executionId = executionId)
             }
         } catch (ex: Exception) {
@@ -168,31 +172,64 @@ class DataBaseReaderGrpcServer(
     }
 
     private class GrpcExecuteListener(
-        private val observer: StreamObserver<QueryResponse>,
+        streamObserver: StreamObserver<QueryResponse>,
+        private val executionId: Long,
         private val event: Event,
         private val onEvent: (Event) -> Unit,
     ) : ResultListener {
-        override fun onRow(sourceId: DataSourceId, row: TableRow) {
-            requireNotNull(row.executionId) {
-                "'Execution id' is null for row: $row"
+        private val channel = Channel<Unit>(1)
+        private val observer = streamObserver as ServerCallStreamObserver<QueryResponse>
+
+        init {
+            observer.setOnReadyHandler {
+                channel.trySend(Unit)
+            }
+            observer.setOnCancelHandler {
+                LOGGER.warn { "gRPC request is canceled for '$executionId' execution" }
+                channel.close()
+            }
+        }
+
+        override suspend fun onRow(sourceId: DataSourceId, row: TableRow) {
+            check(executionId == row.executionId) {
+                "'Execution id' isn't equal to '$executionId', row: $row"
+            }
+            try {
+                if (observer.isCancelled) {
+                    return
+                }
+                if (!observer.isReady) {
+                    channel.receive()
+                }
+
+                observer.onNext(
+                    QueryResponse.newBuilder()
+                        .putRows(row)
+                        .setExecutionId(row.executionId)
+                        .build()
+                )
+            } catch (e: RuntimeException) {
+                if (e is ClosedReceiveChannelException || e is StatusRuntimeException) {
+                    LOGGER.error(e) { "Couldn't send next gRPC message by gRPC connection problem for '$executionId' execution" }
+                } else {
+                    throw e
+                }
             }
-            observer.onNext(
-                QueryResponse.newBuilder()
-                    .putRows(row)
-                    .setExecutionId(row.executionId)
-                    .build()
-            )
         }
 
         override fun onError(error: Throwable) {
-            observer.onError(error)
+            if (!observer.isCancelled) {
+                observer.onError(error)
+            }
             event.endTimestamp()
                 .exception(error, true)
                 .also(onEvent)
         }
 
         override fun onComplete() {
-            observer.onCompleted()
+            if (!observer.isCancelled) {
+                observer.onCompleted()
+            }
             event.endTimestamp()
                 .also(onEvent)
         }
@@ -206,7 +243,7 @@ class DataBaseReaderGrpcServer(
     ) : ResultListener {
         private val counter = AtomicLong()
 
-        override fun onRow(sourceId: DataSourceId, row: TableRow) {
+        override suspend fun onRow(sourceId: DataSourceId, row: TableRow) {
             check(report.executionId == row.executionId) {
                 "'Execution id' isn't equal to '${report.executionId}', row: $row"
             }