0

我正在使用以下方法在 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只是想确保主题出现在列表中。请问这是正确的方法还是任何其他建议？

4

1 回答 1

1

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

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

configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "bootstrap-server-ip:9092");
于 2020-02-28T23:00:54.450 回答