adding retryable logic on consumer, working on #25

Author: juan-medina

command-producer-consumer/pom.xml

@@ -92,6 +92,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>4.0.3</version>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 
     <build>

command-producer-consumer/src/main/kotlin/org/learning/by/example/petstore/command/consumer/CommandsConsumer.kt

@@ -2,7 +2,8 @@ package org.learning.by.example.petstore.command.consumer
 
 import org.learning.by.example.petstore.command.Command
 import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
 
 interface CommandsConsumer {
-    fun receiveCommands(): Flux<Command>
+    fun receiveCommands(sink: (Command) -> Mono<Void>): Flux<Void>
 }

command-producer-consumer/src/main/kotlin/org/learning/by/example/petstore/command/consumer/CommandsConsumerConfig.kt

@@ -15,7 +15,9 @@ data class CommandsConsumerConfig(
     val clientId: String,
     val groupId: String,
     val offsetEarliest: String,
-    val timeoutMS: Int
+    val timeoutMS: Int,
+    val retries: CommandsConsumerConfigRetries
+
 ) {
     companion object Constants {
         const val CONSUMER_CONFIG_PREFIX = "service.commands.consumer"

command-producer-consumer/src/main/kotlin/org/learning/by/example/petstore/command/consumer/CommandsConsumerConfigRetries.kt

@@ -0,0 +1,3 @@
+package org.learning.by.example.petstore.command.consumer
+
+data class CommandsConsumerConfigRetries(val times: Long, val ms: Long)

command-producer-consumer/src/main/kotlin/org/learning/by/example/petstore/command/consumer/CommandsConsumerImpl.kt

@@ -7,11 +7,14 @@ import org.apache.kafka.common.KafkaException
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.learning.by.example.petstore.command.Command
 import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
 import reactor.kafka.receiver.KafkaReceiver
 import reactor.kafka.receiver.ReceiverOptions
 import reactor.kotlin.core.publisher.toMono
+import reactor.util.retry.Retry
+import java.time.Duration
 
-class CommandsConsumerImpl(commandsConsumerConfig: CommandsConsumerConfig, objectMapper: ObjectMapper) :
+class CommandsConsumerImpl(private val commandsConsumerConfig: CommandsConsumerConfig, objectMapper: ObjectMapper) :
     CommandsConsumer {
     private val options = hashMapOf<String, Any>(
         ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to commandsConsumerConfig.bootstrapServer,
@@ -24,6 +27,8 @@ class CommandsConsumerImpl(commandsConsumerConfig: CommandsConsumerConfig, objec
         CommandsDeserializer.OBJECT_MAPPER_CONFIG_KEY to objectMapper
     )
 
+    private val scheduler = Schedulers.newSingle("sample", true)
+
     private val receiver: KafkaReceiver<String, Command> = KafkaReceiver.create(
         ReceiverOptions.create<String, Command>(options).subscription(setOf(commandsConsumerConfig.topic))
     )
@@ -34,15 +39,22 @@ class CommandsConsumerImpl(commandsConsumerConfig: CommandsConsumerConfig, objec
     // To avoid this we will ask first if we can connect to Kafka. However if we can connect, and them subscribed,
     // we may get disconnected latter, but the KafkaConsumer will handle that recovery automatically and we do not
     // need to check it in the future connections.
-    override fun receiveCommands() = isKafkaAvailable.flatMapMany {
-        receiver.receive().flatMap {
-            val receiverOffset = it.receiverOffset()
-            receiverOffset.acknowledge()
-            it.value().toMono()
-        }
+    override fun receiveCommands(sink: (Command) -> Mono<Void>) = isKafkaAvailable.flatMapMany {
+        receiver.receive()
+            .publishOn(scheduler)
+            .concatMap {
+                sink(it.value()).thenEmpty(it.receiverOffset().commit())
+            }.retryWhen(
+                Retry.backoff(
+                    commandsConsumerConfig.retries.times,
+                    Duration.ofMillis(commandsConsumerConfig.retries.ms)
+                ).filter {
+                    it !is ErrorDeserializingObject
+                }
+            )
     }
 
-    // To known if we can't connect to Kafka we could't just get the topics list, only during a set of configurable
+    // To known if we can't connect to Kafka we couldn't just get the topics list, only during a set of configurable
     // time, if we can't get a response in that time we will return an error.
     //
     // We don't need to check what topics they are since we are not really interested in them.

command-producer-consumer/src/test/kotlin/org/learning/by/example/petstore/command/consumer/CommandsConsumerImplTest.kt

@@ -1,29 +1,35 @@
 package org.learning.by.example.petstore.command.consumer
 
 import org.assertj.core.api.Assertions.assertThat
+import org.awaitility.Awaitility.await
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.MethodOrderer
 import org.junit.jupiter.api.Order
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.TestMethodOrder
+import org.learning.by.example.petstore.command.Command
 import org.learning.by.example.petstore.command.consumer.CommandsConsumerConfig.Constants.CONSUMER_CONFIG_BOOSTRAP_SERVER
 import org.learning.by.example.petstore.command.test.CustomKafkaContainer
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.test.annotation.DirtiesContext
 import org.springframework.test.context.ActiveProfiles
 import org.springframework.test.context.DynamicPropertyRegistry
 import org.springframework.test.context.DynamicPropertySource
 import org.testcontainers.junit.jupiter.Container
 import org.testcontainers.junit.jupiter.Testcontainers
+import reactor.core.publisher.Mono
 import reactor.test.StepVerifier
 import java.time.Duration
 import java.time.Instant
 import java.util.UUID
+import java.util.concurrent.TimeUnit
 
 @SpringBootTest
 @Testcontainers
 @ActiveProfiles("consumer")
 @TestMethodOrder(MethodOrderer.OrderAnnotation::class)
+@DirtiesContext
 internal class CommandsConsumerImplTest(
     @Autowired val commandsConsumerImpl: CommandsConsumerImpl,
     @Autowired val commandsConsumerConfig: CommandsConsumerConfig
@@ -79,34 +85,44 @@ internal class CommandsConsumerImplTest(
         assertThat(KAFKA_CONTAINER.sendMessage(commandsConsumerConfig.topic, FIRST_COMMAND)).isTrue()
         assertThat(KAFKA_CONTAINER.sendMessage(commandsConsumerConfig.topic, SECOND_COMMAND)).isTrue()
 
-        StepVerifier.create(commandsConsumerImpl.receiveCommands())
-            .expectSubscription()
-            .thenRequest(Long.MAX_VALUE)
-            .consumeNextWith {
-                assertThat(it.commandName).isEqualTo("example command 1")
-                assertThat(it.id).isEqualTo(UUID.fromString("4cb5294b-1034-4bc4-9b3d-899adb782d89"))
-                assertThat(it.timestamp).isEqualTo(Instant.parse("2020-06-28T08:53:35.168283Z"))
-                assertThat(it.get<String>("attribute1")).isEqualTo("value1")
-                assertThat(it.get<Int>("attribute2")).isEqualTo(123)
-            }
-            .consumeNextWith {
-                assertThat(it.commandName).isEqualTo("example command 2")
-                assertThat(it.id).isEqualTo(UUID.fromString("4cb5294b-1034-4bc4-9b3d-542adb232a21"))
-                assertThat(it.timestamp).isEqualTo(Instant.parse("2020-06-28T11:22:13.456732Z"))
-                assertThat(it.get<Boolean>("attribute1")).isEqualTo(false)
-                assertThat(it.get<Double>("attribute2")).isEqualTo(125.5)
-            }
-            .expectNextCount(0L)
-            .thenCancel()
-            .verify(Duration.ofSeconds(5L))
+        val commands = arrayListOf<Command>()
+
+        val disposable = commandsConsumerImpl.receiveCommands {
+            commands.add(it)
+            Mono.empty()
+        }.subscribe()
+
+        await().atMost(30, TimeUnit.SECONDS).until {
+            commands.size == 2
+        }
+
+        disposable.dispose()
+
+        assertThat(commands).hasSize(2)
+
+        with(commands[0]) {
+            assertThat(commandName).isEqualTo("example command 1")
+            assertThat(id).isEqualTo(UUID.fromString("4cb5294b-1034-4bc4-9b3d-899adb782d89"))
+            assertThat(timestamp).isEqualTo(Instant.parse("2020-06-28T08:53:35.168283Z"))
+            assertThat(get<String>("attribute1")).isEqualTo("value1")
+            assertThat(get<Int>("attribute2")).isEqualTo(123)
+        }
+
+        with(commands[1]) {
+            assertThat(commandName).isEqualTo("example command 2")
+            assertThat(id).isEqualTo(UUID.fromString("4cb5294b-1034-4bc4-9b3d-542adb232a21"))
+            assertThat(timestamp).isEqualTo(Instant.parse("2020-06-28T11:22:13.456732Z"))
+            assertThat(get<Boolean>("attribute1")).isEqualTo(false)
+            assertThat(get<Double>("attribute2")).isEqualTo(125.5)
+        }
     }
 
     @Test
     @Order(1)
     fun `we should error on invalid command`() {
         assertThat(KAFKA_CONTAINER.sendMessage(commandsConsumerConfig.topic, INVALID_COMMAND)).isTrue()
 
-        StepVerifier.create(commandsConsumerImpl.receiveCommands())
+        StepVerifier.create(commandsConsumerImpl.receiveCommands { Mono.empty() })
             .expectSubscription()
             .thenRequest(Long.MAX_VALUE)
             .expectErrorMatches {
@@ -120,7 +136,7 @@ internal class CommandsConsumerImplTest(
     fun `we should error on empty command`() {
         assertThat(KAFKA_CONTAINER.sendMessage(commandsConsumerConfig.topic, EMPTY_COMMAND)).isTrue()
 
-        StepVerifier.create(commandsConsumerImpl.receiveCommands())
+        StepVerifier.create(commandsConsumerImpl.receiveCommands { Mono.empty() })
             .expectSubscription()
             .thenRequest(Long.MAX_VALUE)
             .expectErrorMatches {
@@ -140,7 +156,7 @@ internal class CommandsConsumerImplTest(
 
     @Order(2)
     @Test
-    fun `we can get check if noy alive`() {
+    fun `we can get check if not alive`() {
         if (KAFKA_CONTAINER.isRunning) KAFKA_CONTAINER.stop()
 
         StepVerifier.create(commandsConsumerImpl.isKafkaAvailable)
@@ -156,7 +172,7 @@ internal class CommandsConsumerImplTest(
     fun `we should error on failure to connect`() {
         if (KAFKA_CONTAINER.isRunning) KAFKA_CONTAINER.stop()
 
-        StepVerifier.create(commandsConsumerImpl.receiveCommands())
+        StepVerifier.create(commandsConsumerImpl.receiveCommands { Mono.empty() })
             .expectSubscription()
             .thenRequest(Long.MAX_VALUE)
             .expectErrorMatches {

command-producer-consumer/src/test/resources/application-consumer.yml

@@ -7,3 +7,6 @@ service:
             group-id: pet_commands_consumers
             offset-earliest: earliest
             timeout-ms: 500
+            retries:
+                times: 5
+                ms: 300000

pet-stream/src/main/kotlin/org/learning/by/example/petstore/petstream/service/sink/CommandSink.kt

@@ -13,7 +13,7 @@ class CommandSink(val commandsConsumer: CommandsConsumer, val commandHandler: Co
 
     @PostConstruct
     fun run() {
-        disposable = commandsConsumer.receiveCommands().flatMap(commandHandler::handle).subscribe()
+        disposable = commandsConsumer.receiveCommands(commandHandler::handle).subscribe()
     }
 
     @PreDestroy

pet-stream/src/main/resources/application.yml

@@ -12,5 +12,9 @@ service:
             group-id: pet_commands_consumers
             offset-earliest: earliest
             timeout-ms: 500
+            retries:
+                times: 5
+                ms: 300000
+
 db:
     initialize: false

pet-stream/src/test/kotlin/org/learning/by/example/petstore/petstream/service/sink/CommandSinkTest.kt

@@ -7,14 +7,12 @@ import com.nhaarman.mockitokotlin2.verify
 import com.nhaarman.mockitokotlin2.whenever
 import org.junit.jupiter.api.Test
 import org.learning.by.example.petstore.command.consumer.CommandsConsumer
-import org.learning.by.example.petstore.command.dsl.command
 import org.learning.by.example.petstore.petstream.service.handler.CommandHandler
 import org.learning.by.example.petstore.petstream.test.BasicTest
 import org.springframework.boot.test.context.SpringBootTest
 import org.springframework.boot.test.mock.mockito.MockBean
 import reactor.core.publisher.Mono
 import reactor.kotlin.core.publisher.toFlux
-import java.time.LocalDateTime
 
 @SpringBootTest
 class CommandSinkTest : BasicTest() {
@@ -26,30 +24,14 @@ class CommandSinkTest : BasicTest() {
 
     @Test
     fun `sink should sent to handler commands`() {
-        val cmd1 = command("pet_create") {
-            "name" value "name1"
-            "category" value "category"
-            "breed" value "breed"
-            "vaccines" values listOf("vaccine1", "vaccine2")
-            "dob" value LocalDateTime.now()
-            "tags" values listOf("tag1", "tag2", "tag3")
-        }
-        val cmd2 = command("pet_create") {
-            "name" value "name2"
-            "category" value "category"
-            "breed" value "breed"
-            "vaccines" values listOf("vaccine2")
-            "dob" value LocalDateTime.now()
-            "tags" values listOf("tag1", "tag3")
-        }
-
-        doReturn(listOf(cmd1, cmd2).toFlux()).whenever(consumer).receiveCommands()
+        doReturn(listOf(Mono.empty<Void>(), Mono.empty<Void>()).toFlux()).whenever(consumer).receiveCommands(any())
         doReturn(Mono.empty<Void>()).whenever(handler).handle(any())
 
-        CommandSink(consumer, handler).run()
+        val sink = CommandSink(consumer, handler)
+
+        sink.run()
+        sink.end()
 
-        verify(handler, times(2)).handle(any())
-        verify(handler, times(1)).handle(cmd1)
-        verify(handler, times(1)).handle(cmd2)
+        verify(consumer, times(1)).receiveCommands(handler::handle)
     }
 }

pet-stream/src/test/resources/application.yml

@@ -12,6 +12,9 @@ service:
             group-id: pet_commands_consumers
             offset-earliest: earliest
             timeout-ms: 500
+            retries:
+                times: 5
+                ms: 300000
         producer:
             bootstrap-server: localhost:9092
             topic: pet-commands