Daniel-Jacob

@JvmInline
value class PaymentMethod(val paymentMethod: String)

@JvmInline
value class PaymentId(val paymentId: UUID)

data class Payment(
    val paymentId: PaymentId,
    val paymentMethod: PaymentMethod
)

Daniel-Jacob

val Any.log: Logger get() = LoggerFactory.getLogger(this::class.java)

Daniel-Jacob

object ErrorHandler {
    fun handleError(error: Error) = when(error) {
        is ValidationError -> log.error("a validation error has occured")
        is IOError -> log.error("an IO error has occured")
        is FatalError -> log.error("a Fatal error has occured... exiting")
    }
}

Daniel-Jacob

sealed class Error {
    data class ValidationError(val id: Long, val message: String): Error()
    data class IOError(val message: String, val throwable: Throwable): Error()
    object FatalError : Error()
}

Daniel-Jacob

import org.apache.kafka.common.errors.SerializationException
import org.slf4j.LoggerFactory
import org.springframework.context.annotation.Profile
import org.springframework.kafka.annotation.KafkaListener
import org.springframework.kafka.annotation.RetryableTopic
import org.springframework.kafka.retrytopic.DltStrategy
import org.springframework.kafka.retrytopic.TopicSuffixingStrategy
import org.springframework.kafka.support.converter.ConversionException
import org.springframework.kafka.support.serializer.DeserializationException
import org.springframework.messaging.converter.MessageConversionException

Daniel-Jacob

import org.springframework.boot.autoconfigure.kafka.KafkaProperties
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.context.annotation.Profile
import org.springframework.kafka.annotation.EnableKafka
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
import org.springframework.kafka.core.ConsumerFactory
import org.springframework.kafka.core.DefaultKafkaConsumerFactory
import org.springframework.kafka.core.DefaultKafkaProducerFactory
import org.springframework.kafka.core.KafkaTemplate

Daniel-Jacob

topic: my-topic
retry-attempts: 8
server:
  port: 8080
management:
  server.port: 8081
spring:
  kafka:
    bootstrap-servers: http://localhost:29092
    properties:

Daniel-Jacob

    @Bean
    fun errorHandler(kafkaProperties: KafkaProperties): DefaultErrorHandler =
        run {
            val backOff = ExponentialBackOff(3000, 2.0)
            backOff.maxElapsedTime = 60000
            DefaultErrorHandler(DeadLetterPublishingRecoverer(errorKafkaTemplate(kafkaProperties)), backOff)
        }

Daniel-Jacob

@Component
class FixedBackOffKafkaListener {
    private val log = LoggerFactory.getLogger(FixedBackOffKafkaListener::class.java)

    @KafkaListener(
        id = "\${spring.kafka.consumer.group-id}",
        topics = ["\${topic}"]
    )
    fun onReceive(message: String) {
        log.info("processing message: $message")

Daniel-Jacob

import org.springframework.boot.autoconfigure.kafka.KafkaProperties
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.context.annotation.Profile
import org.springframework.kafka.annotation.EnableKafka
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
import org.springframework.kafka.core.ConsumerFactory
import org.springframework.kafka.core.DefaultKafkaConsumerFactory
import org.springframework.kafka.core.DefaultKafkaProducerFactory
import org.springframework.kafka.core.KafkaTemplate