Kafka Topic单元测试报错咨询：地址分配失败问题排查

First off, your current implementation has a couple of critical issues that are causing the BindException and preventing proper topic validation:

Key Problems in Your Code #

1. Unnecessary Manual Embedded Broker Instance
   You're using the @EmbeddedKafka annotation which already spins up a fully initialized embedded Kafka broker, but then you manually create a new EmbeddedKafkaBroker(2) in your setUp method. This uninitialized broker returns an invalid address (127.0.0.1:0), which is why your AdminClient can't connect.

2. Hardcoded Bootstrap Servers
   Your KafkaTopicAdmin class has a hardcoded "bootstrap-ip" value for bootstrap servers. Even if your test's AdminClient pointed to the right embedded broker, the KafkaTopicAdmin itself would still try to connect to a non-existent server instead of the embedded one.

3. Unwaited Asynchronous Topic Creation
   The createTopics method runs asynchronously, but you don't wait for it to complete before checking for the topic. This could lead to false negatives where your validation runs before the topic is actually created.

Fixed Implementation #

Step 1: Make KafkaTopicAdmin Configurable #

Refactor KafkaTopicAdmin to accept bootstrap servers via constructor injection, so you can point it to the embedded broker during tests:

public class KafkaTopicAdmin { 
    private final String bootstrapServers;

    // Constructor injection for flexible bootstrap server configuration
    public KafkaTopicAdmin(String bootstrapServers) {
        this.bootstrapServers = bootstrapServers;
    }

    public void createTopic(final String topicName, int numPartitions, short replicationFactor) { 
        // Use try-with-resources to auto-close AdminClient and avoid leaks
        try (AdminClient client = getKafkaClient()) { 
            NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor); 
            // Wait for topic creation to finish to avoid race conditions
            client.createTopics(Collections.singletonList(newTopic)).all().get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException("Failed to create topic: " + topicName, e);
        }
    }

    private AdminClient getKafkaClient() { 
        Map<String, Object> configs = new HashMap<>(); 
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); 
        return AdminClient.create(configs); 
    } 
}

Step 2: Fix the Test Class #

Use the embedded broker provided by @EmbeddedKafka instead of creating a new one, and ensure both KafkaTopicAdmin and the test AdminClient use the same valid bootstrap address:

@EmbeddedKafka(partitions = 2, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"})
public class KafkaTopicAdminTest { 
    @Autowired
    private EmbeddedKafkaBroker embeddedKafkaBroker;

    private KafkaTopicAdmin kafkaTopicAdmin; 
    private AdminClient kafkaAdminClient; 

    @Before
    public void setUp(){ 
        // Get the valid bootstrap address from the autowired embedded broker
        String bootstrapServers = embeddedKafkaBroker.getBrokersAsString();
        
        // Initialize KafkaTopicAdmin with the embedded broker's address
        kafkaTopicAdmin = new KafkaTopicAdmin(bootstrapServers); 

        // Initialize test AdminClient with the same address
        Properties properties = new Properties(); 
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); 
        kafkaAdminClient = AdminClient.create(properties); 
    } 

    @Test 
    public void shouldCreateTopicSuccessfully() throws ExecutionException, InterruptedException { 
        String testTopic = "TestTopic";
        // Create topic with explicit partitions/replication factor matching the embedded broker
        kafkaTopicAdmin.createTopic(testTopic, 2, (short)1); 

        // List all topics and verify TestTopic exists
        ListTopicsOptions listTopicsOptions = new ListTopicsOptions(); 
        listTopicsOptions.listInternal(true); 
        Set<String> topicNames = kafkaAdminClient.listTopics(listTopicsOptions).names().get(); 

        // Use assertions instead of print statements for proper test validation
        assertTrue("TestTopic should be present in Kafka", topicNames.contains(testTopic));
    }

    @After
    public void tearDown() {
        // Clean up AdminClient to avoid resource leaks
        if (kafkaAdminClient != null) {
            kafkaAdminClient.close();
        }
    }
}

Additional Optimization Tips #

• Use Configuration Properties: In production, inject bootstrap servers via a configuration class (like @ConfigurationProperties) instead of hardcoding or passing via constructor directly.
• Parameterize Partition/Replication Counts: Pass these values as parameters or inject them from config, just like bootstrap servers, to make the class more flexible.
• Test Containers for Realistic Tests: For more robust integration tests, consider using Testcontainers to spin up a real Kafka cluster instead of relying on embedded Kafka.
• Add Retry Logic: In production, add retry logic for topic creation to handle transient failures gracefully.

内容的提问来源于stack exchange，提问作者parrotjack