import org.springframework.kafka.test.rule.KafkaEmbedded; //导入依赖的package包/类
@SuppressWarnings("serial")
public static KafkaEmbedded kafkaBroker(int port, int partitionCount, String logDirectory, String... topics) {
    final int brokerCount = 1;
    final boolean controlledShutdown = false;

    LOG.debug("JUnit @Rule instantiating embedded Kafka broker: port [{}], topic(s) [{}], brokerCount [{}], partitionCount [{}], controlledShutdown [{}], log directory [{}]",
            port, of(topics).collect(joining(", ")), brokerCount, partitionCount, controlledShutdown, logDirectory);
    KafkaEmbedded embedded = new KafkaEmbedded(brokerCount, controlledShutdown, partitionCount, topics);

    embedded.brokerProperties(new HashMap<String, String>() {
        {
            put("logs.dir", logDirectory);
        }
    });

    embedded.setKafkaPorts(port);

    LOG.debug("JUnit @Rule returning embedded Kafka broker instance [{}]", embedded);
    return embedded;
}
 

import org.springframework.kafka.test.rule.KafkaEmbedded; //导入依赖的package包/类
private static KafkaEmbedded buildEmbeddedKafka() {
        //Build a list of all the topics to create and along thw way create a map for each of
        //the topic types
        Arrays.stream(StatisticType.values())
                .forEachOrdered(type -> {
                    String inputTopic = TopicNameFactory.getStatisticTypedName(STATISTIC_EVENTS_TOPIC_PREFIX, type);
                    topics.add(inputTopic);
                    INPUT_TOPICS_MAP.put(type, inputTopic);

                    String badTopic = TopicNameFactory.getStatisticTypedName(BAD_STATISTIC_EVENTS_TOPIC_PREFIX, type);
                    topics.add(badTopic);
                    BAD_TOPICS_MAP.put(type, badTopic);

                    Arrays.stream(EventStoreTimeIntervalEnum.values())
                            .forEachOrdered(interval -> {
                                String rollupTopic = TopicNameFactory.getIntervalTopicName(
                                        STATISTIC_ROLLUP_PERMS_TOPIC_PREFIX,
                                        type,
                                        interval);
                                topics.add(rollupTopic);
                                ROLLUP_TOPICS_MAP.computeIfAbsent(type, k -> new ArrayList<>()).add(rollupTopic);
                            });
                });


//        topics.forEach(topic -> LOGGER.info("Creating topic: {}", topic));

//        return new KafkaEmbedded(1, true, 1, topics.toArray(new String[topics.size()]));
        return new KafkaEmbedded(1, true, 1);
    }
 

import org.springframework.kafka.test.rule.KafkaEmbedded; //导入依赖的package包/类
public static void createTopics(final KafkaEmbedded kafkaEmbedded, final String... topics) {

        ZkUtils zkUtils = new ZkUtils(kafkaEmbedded.getZkClient(), null, false);
        for (String topic : topics) {
            LOGGER.info("Creating topic {}", topic);
            AdminUtils.createTopic(zkUtils,
                    topic,
                    kafkaEmbedded.getPartitionsPerTopic(),
                    kafkaEmbedded.getBrokerAddresses().length,
                    new Properties(),
                    null);
        }
    }
 

import org.springframework.kafka.test.rule.KafkaEmbedded; //导入依赖的package包/类
public static void deleteTopics(final KafkaEmbedded kafkaEmbedded, final String... topics) {
    ZkUtils zkUtils = new ZkUtils(kafkaEmbedded.getZkClient(), null, false);
    for (String topic : topics) {

        if (AdminUtils.topicExists(zkUtils, topic)) {
            LOGGER.info("Deleting topic {}", topic);
            kafkaEmbedded.waitUntilSynced(topic, 0);
            AdminUtils.deleteTopic(zkUtils, topic);
            while (AdminUtils.topicExists(zkUtils, topic)) {
                //wait for topic to die
            }
            kafkaEmbedded.waitUntilSynced(topic, 0);
        }
    }
}
 

import org.springframework.kafka.test.rule.KafkaEmbedded; //导入依赖的package包/类
@Bean
KafkaEmbedded kafka() {

	KafkaEmbedded embedded = new KafkaEmbedded(1);
	embedded.setKafkaPorts(9092);
	return embedded;
}
 

import org.springframework.kafka.test.rule.KafkaEmbedded; //导入依赖的package包/类
public static void deleteAndCreateTopics(final KafkaEmbedded kafkaEmbedded, final String... topics) {
    //attempt deletion first to ensure the topics don't already exist
    deleteTopics(kafkaEmbedded, topics);
    createTopics(kafkaEmbedded, topics);
}