@Override protected void before() throws Throwable {
    // Start the ZK and the Broker
    LOG.info("init embedded Zookeeper");
    tmpLogDir = Files.createTempDirectory("kafka-log-dir-").toAbsolutePath();
    zkServer = new EmbeddedZookeeper();
    String zkConnect = IPStackUtils.concatLoopbackAddressPort(zkServer.port());
    LOG.info("init kafka broker");
    Properties brokerProps = new Properties();
    brokerProps.setProperty("zookeeper.connect", zkConnect);
    brokerProps.setProperty("broker.id", "0");
    brokerProps.setProperty("log.dir", tmpLogDir.toString());
    Map<String, BrokerListener> listeners = new HashMap<>();
    listeners.put("L1", new BrokerListener(BROKER_IP_PORT, "PLAINTEXT"));
    if (principal != null) {
      listeners.put("L2", new BrokerListener(BROKER_SASL_PORT, "SASL_PLAINTEXT"));
      listeners.put("L3", new BrokerListener(BROKER_SASL_SSL_PORT, "SASL_SSL"));
    }
    String listenersURLs = listeners.entrySet().stream().map((e) -> e.getKey() + "://" + e.getValue().url)
        .collect(Collectors.joining(","));
    brokerProps.setProperty("listeners", listenersURLs);
    String listenersProtocols = listeners.entrySet().stream().map((e) -> e.getKey() + ":" + e.getValue().protocol)
        .collect(Collectors.joining(","));
    brokerProps.setProperty("listener.security.protocol.map", listenersProtocols);
    brokerProps.setProperty("inter.broker.listener.name", "L1");
    if (principal != null) {
      String jaasConfig = String.format("%s %s %s %s serviceName=\"%s\" keyTab=\"%s\" principal=\"%s\";",
          "com.sun.security.auth.module.Krb5LoginModule required", "debug=true", "useKeyTab=true", "storeKey=true",
          principal, keytab, principal + "/localhost");
      brokerProps.setProperty("listener.name.l2.gssapi.sasl.jaas.config", jaasConfig);
      brokerProps.setProperty("listener.name.l3.gssapi.sasl.jaas.config", jaasConfig);
      truststoreFile = File.createTempFile("kafka_truststore", "jks");
      brokerProps.putAll(new TestSslUtils.SslConfigsBuilder(Mode.SERVER).createNewTrustStore(truststoreFile).build());
      brokerProps.setProperty("delegation.token.master.key", "AnyValueShouldDoHereItDoesntMatter");
    }
    brokerProps.setProperty("offsets.topic.replication.factor", "1");
    brokerProps.setProperty("transaction.state.log.replication.factor", "1");
    brokerProps.setProperty("transaction.state.log.min.isr", "1");
    KafkaConfig config = new KafkaConfig(brokerProps);
    kafkaServer = TestUtils.createServer(config, Time.SYSTEM);
    kafkaServer.startup();
    kafkaServer.zkClient();
    adminZkClient = new AdminZkClient(kafkaServer.zkClient());
    LOG.info("Creating kafka TOPIC [{}]", TOPIC);
    adminZkClient.createTopic(TOPIC, 1, 1, new Properties(), RackAwareMode.Disabled$.MODULE$);
  }
Override to set up your specific external resource.

@throws Throwable if setup fails (which will disable {@code after}
kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaBrokerResource.java