readStringMessages takes 30+ seconds about kafka-junit HOT 9 CLOSED

import com.github.charithe.kafka.KafkaJunitRule;
import com.google.common.collect.ImmutableMap;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.Rule;
import org.junit.Test;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;

import static org.junit.Assert.assertEquals;

public class KafkaJUnitTestRule {
    @Rule public KafkaJunitRule kafka = new KafkaJunitRule();

    @Test
    public void test() throws TimeoutException, IOException, ExecutionException, InterruptedException {
        final Map<String, Object> producerProperties = ImmutableMap.<String, Object>builder()
            .put("bootstrap.servers", "127.0.0.1:" + kafka.kafkaBrokerPort())
            .put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
            .put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
            .put("request.required.acks", "1")
            .build();

        final Producer<String, String> producer = new KafkaProducer<>(producerProperties);

        final ProducerRecord<String, String> record = new ProducerRecord<>("roger", "rogercyr", "Roger Cyr is the best printer administrator");

        final Future<RecordMetadata> future = producer.send(record);
        final RecordMetadata recordMetadata = future.get();
        producer.close();

        final List<String> messages = kafka.readStringMessages("roger", 1);
        final String message = messages.get(0);
        assertEquals("Roger Cyr is the best printer administrator", message);
    }
}

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.9.0.1</version>
        </dependency>
        <dependency>
            <groupId>com.github.charithe</groupId>
            <artifactId>kafka-junit</artifactId>
            <version>1.9</version>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>log4j-over-slf4j</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

  def shutdown() {
    val canShutdown = isShuttingDown.compareAndSet(false, true);
    if (canShutdown) {
      logger.info("ZKConsumerConnector shutting down")
      try {
        scheduler.shutdown
        fetcher match {
          case Some(f) => f.shutdown
          case None =>
        }
        sendShudownToAllQueues
        if (zkClient != null) {
          zkClient.close()
          zkClient = null
        }
      }
      catch {
        case e =>
          logger.fatal(e)
          logger.fatal(Utils.stackTrace(e))
      }
      logger.info("ZKConsumerConnector shut down completed")
    }
  }