refactor(clients): remove unnecessary object classes

fhussonnois · fhussonnois · commit ad982accc036 · 2020-08-05T23:03:29.000+02:00
diff --git a/README.md b/README.md
@@ -163,7 +163,7 @@ val consumerWorker: ConsumerWorker<String, String> = kafka("localhost:9092") {
             autoOffsetReset(AutoOffsetReset.Earliest)
         }
 
-        onDeserializationError(silentlyReplaceWithNull())
+        onDeserializationError(replaceWithNullOnInvalidRecord())
 
         onPartitionsAssigned { _: Consumer<*, *>, partitions ->
             println("Partitions assigned: $partitions")
diff --git a/clients/src/main/kotlin/io/streamthoughts/kafka/clients/consumer/KafkaConsumerWorker.kt b/clients/src/main/kotlin/io/streamthoughts/kafka/clients/consumer/KafkaConsumerWorker.kt
@@ -20,9 +20,9 @@ package io.streamthoughts.kafka.clients.consumer
 
 import io.streamthoughts.kafka.clients.consumer.KafkaConsumerWorker.KafkaConsumerWorker
 import io.streamthoughts.kafka.clients.consumer.error.ConsumedErrorHandler
-import io.streamthoughts.kafka.clients.consumer.error.ConsumedErrorHandlers.closeTaskOnConsumedError
+import io.streamthoughts.kafka.clients.consumer.error.closeTaskOnConsumedError
 import io.streamthoughts.kafka.clients.consumer.error.serialization.DeserializationErrorHandler
-import io.streamthoughts.kafka.clients.consumer.error.serialization.DeserializationErrorHandlers
+import io.streamthoughts.kafka.clients.consumer.error.serialization.logAndFailOnInvalidRecord
 import io.streamthoughts.kafka.clients.consumer.listener.ConsumerBatchRecordsListener
 import io.streamthoughts.kafka.clients.consumer.listener.noop
 import io.streamthoughts.kafka.clients.loggerFor
@@ -209,7 +209,7 @@ class KafkaConsumerWorker<K, V> (
                 SimpleConsumerAwareRebalanceListener(),
                 batchRecordListener ?: noop(),
                 onConsumedError ?: closeTaskOnConsumedError(),
-                onDeserializationError ?: DeserializationErrorHandlers.logAndFail(),
+                onDeserializationError ?: logAndFailOnInvalidRecord(),
                 consumerFactory ?: ConsumerFactory.DefaultConsumerFactory
             )
 
diff --git a/clients/src/main/kotlin/io/streamthoughts/kafka/clients/consumer/error/ConsumedErrorHandlers.kt b/clients/src/main/kotlin/io/streamthoughts/kafka/clients/consumer/error/ConsumedErrorHandlers.kt
@@ -27,11 +27,8 @@ import org.slf4j.Logger
 import java.time.Duration
 import kotlin.math.max
 
-
-object ConsumedErrorHandlers {
-    fun closeTaskOnConsumedError(): ConsumedErrorHandler = CloseTaskOnConsumedError
-    fun logAndCommitOnConsumedError(): ConsumedErrorHandler = LogAndCommitOnConsumedError
-}
+fun closeTaskOnConsumedError(): ConsumedErrorHandler = CloseTaskOnConsumedError
+fun logAndCommitOnConsumedError(): ConsumedErrorHandler = LogAndCommitOnConsumedError
 
 /**
  * Stops the [ConsumerTask] when an error is thrown while a non-empty batch of [ConsumerRecord] is being processed
diff --git a/clients/src/main/kotlin/io/streamthoughts/kafka/clients/consumer/error/serialization/DeserializationErrorHandlers.kt b/clients/src/main/kotlin/io/streamthoughts/kafka/clients/consumer/error/serialization/DeserializationErrorHandlers.kt
@@ -21,16 +21,13 @@ package io.streamthoughts.kafka.clients.consumer.error.serialization
 import io.streamthoughts.kafka.clients.loggerFor
 import org.apache.kafka.clients.consumer.ConsumerRecord
 
-object DeserializationErrorHandlers {
+fun <K, V> replaceWithOnInvalidRecord(key: K, value: V): DeserializationErrorHandler<K, V> = ReplaceErrorHandler(key, value)
 
-    fun <K, V> silentlyReplaceWith(key: K, value: V): DeserializationErrorHandler<K, V> = ReplaceErrorHandler(key, value)
+fun <K, V> replaceWithNullOnInvalidRecord(): DeserializationErrorHandler<K, V> = ReplaceErrorHandler()
 
-    fun <K, V> silentlyReplaceWithNull(): DeserializationErrorHandler<K, V> = ReplaceErrorHandler()
+fun <K, V> logAndFailOnInvalidRecord(): DeserializationErrorHandler<K, V> = LogAndFailErrorHandler()
 
-    fun <K, V> logAndFail(): DeserializationErrorHandler<K, V> = LogAndSkipErrorHandler()
-
-    fun <K, V> logAndSkip(): DeserializationErrorHandler<K, V> = LogAndFailErrorHandler()
-}
+fun <K, V> logAndSkipOnInvalidRecord(): DeserializationErrorHandler<K, V> = LogAndSkipErrorHandler()
 
 private class ReplaceErrorHandler<K, V>(
     private val key: K? = null,
diff --git a/clients/src/test/kotlin/io/streamthoughts/kafka/clients/consumer/KafkaConsumerConfigsTest.kt b/clients/src/test/kotlin/io/streamthoughts/kafka/clients/consumer/KafkaConsumerConfigsTest.kt
@@ -18,8 +18,6 @@
  */
 package io.streamthoughts.kafka.clients.consumer
 
-import io.streamthoughts.kafka.clients.Kafka
-import io.streamthoughts.kafka.clients.KafkaClientConfigs
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.junit.jupiter.api.Assertions.assertEquals
@@ -29,9 +27,6 @@ import org.junit.jupiter.api.TestInstance
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class KafkaConsumerConfigsTest {
 
-    private val kafka = Kafka(bootstrapServers = arrayOf("dummy:1234"))
-    private val client = KafkaClientConfigs(kafka = kafka).clientId("clientId")
-
     @Test
     fun should_return_valid_kafka_consumer_config() {
         val configs: KafkaConsumerConfigs = consumerConfigsOf()
diff --git a/clients/src/test/kotlin/io/streamthoughts/kafka/clients/consumer/KafkaConsumerTaskTest.kt b/clients/src/test/kotlin/io/streamthoughts/kafka/clients/consumer/KafkaConsumerTaskTest.kt
@@ -19,8 +19,8 @@
 package io.streamthoughts.kafka.clients.consumer
 
 import io.streamthoughts.kafka.clients.consumer.error.ConsumedErrorHandler
-import io.streamthoughts.kafka.clients.consumer.error.ConsumedErrorHandlers
-import io.streamthoughts.kafka.clients.consumer.error.serialization.DeserializationErrorHandlers
+import io.streamthoughts.kafka.clients.consumer.error.closeTaskOnConsumedError
+import io.streamthoughts.kafka.clients.consumer.error.serialization.logAndFailOnInvalidRecord
 import io.streamthoughts.kafka.clients.consumer.listener.ConsumerBatchRecordsListener
 import io.streamthoughts.kafka.clients.loggerFor
 import io.streamthoughts.kafka.clients.producer.Acks
@@ -121,7 +121,7 @@ class KafkaConsumerTaskTest(private val cluster: TestingEmbeddedKafka) {
 
         val consumer = createConsumerFor(
             listener = failingListener,
-            consumedErrorHandler = ConsumedErrorHandlers.closeTaskOnConsumedError()
+            consumedErrorHandler = closeTaskOnConsumedError()
         )
         runBlocking {
             val job = GlobalScope.launch { consumer.run() }
@@ -147,7 +147,7 @@ class KafkaConsumerTaskTest(private val cluster: TestingEmbeddedKafka) {
     }
 
     private fun createConsumerFor(listener: ConsumerBatchRecordsListener<String, String>,
-                                  consumedErrorHandler: ConsumedErrorHandler = ConsumedErrorHandlers.closeTaskOnConsumedError()
+                                  consumedErrorHandler: ConsumedErrorHandler = closeTaskOnConsumedError()
     ): KafkaConsumerTask<String, String> {
         return KafkaConsumerTask<String, String>(
             consumerFactory = ConsumerFactory.DefaultConsumerFactory,
@@ -157,7 +157,7 @@ class KafkaConsumerTaskTest(private val cluster: TestingEmbeddedKafka) {
             valueDeserializer = StringDeserializer(),
             listener = listener,
             clientId = "test-client",
-            deserializationErrorHandler = DeserializationErrorHandlers.logAndFail(),
+            deserializationErrorHandler = logAndFailOnInvalidRecord(),
             consumedErrorHandler = consumedErrorHandler
         )
     }
diff --git a/clients/src/test/kotlin/io/streamthoughts/kafka/clients/producer/KafkaProducerContainerTest.kt b/clients/src/test/kotlin/io/streamthoughts/kafka/clients/producer/KafkaProducerContainerTest.kt
@@ -36,7 +36,7 @@ import org.junit.jupiter.api.TestInstance
 import org.junit.jupiter.api.extension.ExtendWith
 import org.slf4j.Logger
 import java.time.Duration
-import java.util.*
+import java.util.Properties
 
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 @ExtendWith(EmbeddedSingleNodeKafkaCluster::class)
diff --git a/examples/src/main/kotlin/io/streamthoughts/kafka/client/examples/ConsumerKotlinDSLExample.kt b/examples/src/main/kotlin/io/streamthoughts/kafka/client/examples/ConsumerKotlinDSLExample.kt
@@ -21,8 +21,8 @@ package io.streamthoughts.kafka.client.examples
 import io.streamthoughts.kafka.clients.consumer.AutoOffsetReset
 import io.streamthoughts.kafka.clients.consumer.ConsumerTask
 import io.streamthoughts.kafka.clients.consumer.ConsumerWorker
-import io.streamthoughts.kafka.clients.consumer.error.ConsumedErrorHandlers
-import io.streamthoughts.kafka.clients.consumer.error.serialization.DeserializationErrorHandlers
+import io.streamthoughts.kafka.clients.consumer.error.closeTaskOnConsumedError
+import io.streamthoughts.kafka.clients.consumer.error.serialization.replaceWithNullOnInvalidRecord
 import io.streamthoughts.kafka.clients.consumer.listener.forEach
 import io.streamthoughts.kafka.clients.kafka
 import kotlinx.coroutines.delay
@@ -47,9 +47,9 @@ fun main(args: Array<String>) {
                 autoOffsetReset(AutoOffsetReset.Earliest)
             }
 
-            onDeserializationError(DeserializationErrorHandlers.silentlyReplaceWithNull())
+            onDeserializationError(replaceWithNullOnInvalidRecord())
 
-            onConsumedError(ConsumedErrorHandlers.closeTaskOnConsumedError())
+            onConsumedError(closeTaskOnConsumedError())
 
             onPartitionsAssigned { _: Consumer<*, *>, partitions ->
                 println("Partitions assigned: $partitions")

Original file line number	Diff line number	Diff line change
`@@ -163,7 +163,7 @@ val consumerWorker: ConsumerWorker<String, String> = kafka("localhost:9092") {`
`163`	`163`	`autoOffsetReset(AutoOffsetReset.Earliest)`
`164`	`164`	`}`
`165`	`165`
`166`		`- onDeserializationError(silentlyReplaceWithNull())`
	`166`	`+ onDeserializationError(replaceWithNullOnInvalidRecord())`
`167`	`167`
`168`	`168`	`onPartitionsAssigned { _: Consumer<, >, partitions ->`
`169`	`169`	`println("Partitions assigned: $partitions")`