Kafka主题单元测试

Question

我正在通过以下方法在Kafka中创建主题，

public class  KafkaTopicAdmin {

  public void createTopic(final String topicName) {

        final AdminClient client = getKafkaClient();
        final List<NewTopic> topics = Collections.synchronizedList(new ArrayList<>());
        final NewTopic newTopic = new NewTopic(topicName, connectionConfig.getNoOfPartition(), (short) connectionConfig.getNoOfReplicas());
        topics.add(newTopic);
        client.createTopics(topics);

    }

private AdminClient getKafkaClient() {
        final Map<String, Object> configs = new ConcurrentHashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "bootstrap-ip");
        return AdminClient.create(configs);
    }
}

我的测试班是，

@EmbeddedKafka
public class KafkaTopicAdminTest {

    private KafkaTopicAdmin kafkaTopicAdmin;
    private AdminClient kafkaAdminClient;

    @Before
    public void setUp(){

        kafkaTopicAdmin = new KafkaTopicAdmin();
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,  StringUtils.arrayToCommaDelimitedString(new EmbeddedKafkaBroker(2).getBrokerAddresses()));
        kafkaAdminClient = KafkaAdminClient.create(properties);
    }

    @Test
    public void shouldCreateTopic() throws ExecutionException, InterruptedException {
        kafkaTopicAdmin.createTopic("TestTopic");
        ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
        listTopicsOptions.listInternal(true);
        System.out.println("topics:" + kafkaAdminClient.listTopics(listTopicsOptions).names().get());
    }
}

我收到以下错误，

[AdminClient clientId=adminclient-1] Error connecting to node 127.0.0.1:0 (id: -2 rack: null)
java.net.BindException: Can't assign requested address

我使用@EmbeddedKafka我只是想确保列表中存在该主题。请问这是正确的方法还是其他建议？

Answer 1

从错误看来，您指定的bootstrap.servers属性没有端口：127.0.0.1:0

此属性也占用您的Kafka引导服务器的端口，默认情况下为9092，因此请尝试以下操作：

configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "bootstrap-server-ip:9092");