fix: test grpc status on code

okp4 · Apr 8, 2022 · b3f0347 · b3f0347
1 parent 532646e
commit b3f0347
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/src/main/kotlin/com/okp4/connect/cosmos/CosmosSourceTask.kt b/src/main/kotlin/com/okp4/connect/cosmos/CosmosSourceTask.kt
@@ -6,6 +6,7 @@ import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.runBlocking
 import org.apache.kafka.connect.data.Schema
 import org.apache.kafka.connect.errors.ConnectException
+import org.apache.kafka.connect.errors.RetriableException
 import org.apache.kafka.connect.source.SourceRecord
 import org.apache.kafka.connect.source.SourceTask
 import tendermint.types.BlockOuterClass.Block
@@ -53,7 +54,7 @@ class CosmosSourceTask : SourceTask() {
                     .takeWhile { !serviceClient.isClosed() }
                     .map { serviceClient.getBlockByHeight(it) }
                     .map { it.getOrThrow() }
-                    .catch { if (it is StatusException && it.status.code != Status.INVALID_ARGUMENT.code) throw it }
+                    .catch { if (it is StatusException && it.status.code != Status.INVALID_ARGUMENT.code) throw RetriableException(it) }
                     .map { asSourceRecord(it) }
                     .toList()
             }

diff --git a/src/test/kotlin/com/okp4/connect/cosmos/CosmosSourceTaskTest.kt b/src/test/kotlin/com/okp4/connect/cosmos/CosmosSourceTaskTest.kt
@@ -8,6 +8,7 @@ import io.kotest.datatest.withData
 import io.kotest.matchers.shouldBe
 import io.mockk.*
 import org.apache.kafka.connect.errors.ConnectException
+import org.apache.kafka.connect.errors.RetriableException
 import tendermint.types.BlockOuterClass
 import tendermint.types.Types
 
@@ -115,11 +116,11 @@ class CosmosSourceTaskTest : BehaviorSpec({
             cosmosSourceTask.start(props)
 
             then("it shall complete exceptionally") {
-                val thrown = shouldThrow<StatusException> {
+                val thrown = shouldThrow<RetriableException> {
                     cosmosSourceTask.poll()
                 }
 
-                thrown.status shouldBe Status.DEADLINE_EXCEEDED
+                (thrown.cause as StatusException).status shouldBe Status.DEADLINE_EXCEEDED
 
                 coVerifyOrder {
                     cosmosClient.getBlockByHeight(1)