@@ -11,6 +11,8 @@ import org.apache.kafka.clients.producer.KafkaProducer
1111import org.apache.kafka.clients.producer.ProducerConfig
1212import org.apache.kafka.common.serialization.BytesDeserializer
1313import org.apache.kafka.common.serialization.BytesSerializer
14+ import org.apache.kafka.common.serialization.StringDeserializer
15+ import org.apache.kafka.common.serialization.StringSerializer
1416import org.apache.kafka.common.utils.Bytes
1517import org.testcontainers.containers.KafkaContainer
1618import org.testcontainers.utility.DockerImageName
@@ -43,6 +45,15 @@ fun KafkaContainer.producer(configure: Properties.() -> Unit = {}): KafkaProduce
4345 return KafkaProducer <Bytes , Bytes >(props)
4446}
4547
48+ fun KafkaContainer.stringStringProducer (configure : Properties .() -> Unit = {}): KafkaProducer <String , String > {
49+ val props = Properties ()
50+ props[CommonClientConfigs .BOOTSTRAP_SERVERS_CONFIG ] = bootstrapServers
51+ props[ProducerConfig .KEY_SERIALIZER_CLASS_CONFIG ] = StringSerializer ::class .java
52+ props[ProducerConfig .VALUE_SERIALIZER_CLASS_CONFIG ] = StringSerializer ::class .java
53+ props.configure()
54+ return KafkaProducer <String , String >(props)
55+ }
56+
4657fun KafkaContainer.consumer (configure : Properties .() -> Unit = {}): KafkaConsumer <Bytes , Bytes > {
4758 val props = Properties ()
4859 props[CommonClientConfigs .BOOTSTRAP_SERVERS_CONFIG ] = bootstrapServers
@@ -53,3 +64,14 @@ fun KafkaContainer.consumer(configure: Properties.() -> Unit = {}): KafkaConsume
5364 props.configure()
5465 return KafkaConsumer <Bytes , Bytes >(props)
5566}
67+
68+ fun KafkaContainer.stringStringConsumer (configure : Properties .() -> Unit = {}): KafkaConsumer <String , String > {
69+ val props = Properties ()
70+ props[CommonClientConfigs .BOOTSTRAP_SERVERS_CONFIG ] = bootstrapServers
71+ props[ConsumerConfig .GROUP_ID_CONFIG ] = " kotest_consumer_" + System .currentTimeMillis()
72+ props[ConsumerConfig .AUTO_OFFSET_RESET_CONFIG ] = " earliest"
73+ props[ConsumerConfig .KEY_DESERIALIZER_CLASS_CONFIG ] = StringDeserializer ::class .java
74+ props[ConsumerConfig .VALUE_DESERIALIZER_CLASS_CONFIG ] = StringDeserializer ::class .java
75+ props.configure()
76+ return KafkaConsumer <String , String >(props)
77+ }
0 commit comments