编写测试以在多线程环境中使用 Kafka 消费者

Question

我正在尝试在一个单独的线程中创建一个 kafka 消费者，该线程使用来自 kafka 主题的数据。为此，我扩展了 ShutdownableThread 抽象 class 并提供了 doWork 方法的实现。我的代码是这样的-

abstract class MyConsumer(topic: String) extends ShutdownableThread(topic) {
    val props: Properties = ???
    private val consumer = new KafkaConsumer[String, String](props)
    consumer.subscribe(List(topic).asJava)

    def process(value: String): Unit // Abstract method defining what to do with each record

    override def doWork(): Unit = {
        for (record <- consumer.poll(Duration.ofMillis(1000)).asScala)
            process(record.value())
    }
}

现在在我的测试中，我创建了消费者提供 process() 方法的实现，该方法只是改变一个变量，然后调用它的 start() 方法来启动线程。

var mutVar = "initial_value"

val consumer = new MyConsumer("test_topic") {
    override def process(value: String): Unit = mutVar = "updated_value"
}

consumer.start()
assert(mutVar === "updated_value")

Consumer 确实消费了来自kafka 的消息，但它在测试完成之前没有更新它，因此测试失败。所以，我试图让主线程进入睡眠状态。但它抛出 ConcurrentModificationException 异常消息 - KafkaConsumer is not safe for multi-threaded access

知道我的方法有什么问题吗？提前致谢。

Answer 1

不得不让主线程休眠几秒钟，以允许消费者使用来自 kafka 主题的消息并将其存储在可变变量中。添加 - Thread.sleep(5000) 启动消费者后。

编写测试以在多线程环境中使用 Kafka 消费者

Writing tests to use Kafka consumer in multi-threading environment

multithreading

scala

thread-safety

apache-kafka

kafka-consumer-api