我想测试一个使用kafka消息并将其写入日志的应用程序。下面是它在类似scala的伪代码中的近似表示：

import kafka.consumer.Consumer
import kafka.consumer.ConsumerConfig
import org.slf4j.LoggerFactory
import java.util.Properties
import java.util.HashMap

object ConsumerApp extends App {
  val topic = new HashMap[String, Integer]()
  topic.put("test", 1)
  val logger = LoggerFactory.getLogger(getClass().getName())
  val messageStream = Consumer
    .createJavaConsumerConnector(new ConsumerConfig(new Properties()))
    .createMessageStreams(topic)
    .get(topic).get(0)
  for (message <- messageStream) {
    val gotMessage = new String(message.message())
    logger.info(gotMessage)
  }
}

我想到的测试场景如下：
kafka服务器已启动。
应用程序启动并连接到kafka服务器，开始侦听特定to主题的消息。
将向主题发送消息。
应用程序使用消息并记录它。
下面是用类似scala的伪代码编写的测试草稿：

import uk.org.lidalia.slf4jtest.TestLoggerFactory;
import uk.org.lidalia.slf4jtest.LoggingEvent.info;

abstract class UnitSpec extends FlatSpec with Matchers with EmbeddedKafka {

}

class ConsumerAppSpec extends UnitSpec {
  "ConsumerApp" should "consume and log messages from Kafka on specific topic" in {
    withRunningKafka {
      val consumer = ConsumerApp
      // interecept logger to be able to test that the kafka message is logged
      val logger = TestLoggerFactory.getTestLogger(consumer.getClass)
      // start the application, but beforehand do something to prevent it infinitely blocking
      ???
      consumer.main(Array())  

      // publish a test message    
      publishStringMessageToKafka("test", "TEST")
      // Confirm that the message has been properly logged
      ???
    }
    EmbeddedKafka.stop()
  }
}

我的问题是测试代码的前三个问号。如果我执行main（）方法，它将不会终止，从而阻止执行测试的其余部分。