Introduction

When people ask about getHostnamePort in a Flink plus Kafka setup, they are usually trying to answer a simpler question: what exact host:port value should the Kafka connector use to reach the brokers. In practice, Flink does not need a magic helper method as much as it needs a valid Kafka bootstrap server string.

That string normally looks like broker-1:9092,broker-2:9092. If the hostname is wrong, the listener configuration is wrong, or the port is unreachable from the Flink task managers, the connector fails even though the code compiles cleanly.

What Flink Actually Expects

Kafka clients in Flink use the same fundamental connection setting as regular Kafka clients: bootstrap.servers. The value is a comma-separated list of broker endpoints, not a full URL and not a ZooKeeper address.

With the modern Kafka source API, the configuration looks like this:

1import org.apache.flink.api.common.eventtime.WatermarkStrategy;
2import org.apache.flink.connector.kafka.source.KafkaSource;
3import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
4import org.apache.flink.api.common.serialization.SimpleStringSchema;
5
6KafkaSource<String> source = KafkaSource.<String>builder()
7    .setBootstrapServers("broker-1:9092,broker-2:9092")
8    .setTopics("orders")
9    .setGroupId("flink-orders")
10    .setStartingOffsets(OffsetsInitializer.earliest())
11    .setValueOnlyDeserializer(new SimpleStringSchema())
12    .build();

The important part is the bootstrap string. If you are building that value dynamically, your helper only needs to return correct host:port pairs in the format Kafka expects.

Build the host:port String Safely

A small helper method is often enough when the host and port come from environment variables or application config:

1public final class KafkaAddress {
2    private KafkaAddress() {
3    }
4
5    public static String getHostnamePort(String host, int port) {
6        if (host == null || host.isBlank()) {
7            throw new IllegalArgumentException("host must not be blank");
8        }
9        if (port < 1 || port > 65535) {
10            throw new IllegalArgumentException("port out of range");
11        }
12        return host + ":" + port;
13    }
14
15    public static void main(String[] args) {
16        System.out.println(getHostnamePort("broker-1", 9092));
17    }
18}

For a single broker, that returns a valid endpoint such as broker-1:9092. For a cluster, join several of them with commas:

1String bootstrapServers = String.join(",",
2    KafkaAddress.getHostnamePort("broker-1", 9092),
3    KafkaAddress.getHostnamePort("broker-2", 9092),
4    KafkaAddress.getHostnamePort("broker-3", 9092)
5);

This is not complicated logic, but validating the inputs is worth it. A malformed port or empty host name is much easier to catch here than from a connector timeout later.

Network and Kafka Listener Details Matter More Than the Helper

Many Flink plus Kafka issues are blamed on string formatting when the real problem is Kafka broker listener configuration. A broker can accept the initial bootstrap connection and still return advertised addresses that the Flink workers cannot reach.

That means you should verify:

• the host names resolve from every Flink task manager
• the advertised Kafka listener points to reachable host names and ports
• the correct security protocol is configured when TLS or SASL is enabled

A quick Java connectivity check can help isolate network issues:

1import java.net.InetSocketAddress;
2import java.net.Socket;
3
4public class PortCheck {
5    public static void main(String[] args) throws Exception {
6        try (Socket socket = new Socket()) {
7            socket.connect(new InetSocketAddress("broker-1", 9092), 3000);
8            System.out.println("reachable");
9        }
10    }
11}

If this fails from the same network where Flink runs, the problem is not inside Flink code.

Prefer Config Objects Over Hardcoding

Hardcoding broker endpoints in source code is fine for a tutorial but weak in real deployments. A better pattern is to read them from environment variables, a properties file, or your deployment platform:

1String bootstrapServers = System.getenv("KAFKA_BOOTSTRAP_SERVERS");
2if (bootstrapServers == null || bootstrapServers.isBlank()) {
3    throw new IllegalStateException("KAFKA_BOOTSTRAP_SERVERS is missing");
4}

That makes it possible to keep the same Flink job binary across local development, staging, and production.

Common Pitfalls

The first pitfall is confusing Kafka broker endpoints with HTTP URLs. bootstrap.servers expects host:port entries, not values like http://broker-1:9092.

Another common issue is using the right host name from your laptop but the wrong one from inside the Flink cluster. Containers, Kubernetes services, and cloud DNS names often differ from local development assumptions.

People also forget that Kafka may advertise a different address than the one used for the initial connection. When that advertised listener is wrong, the connector appears to connect and then fails on subsequent broker communication.

Finally, do not overcomplicate the code. A getHostnamePort helper is useful, but the real operational work is validating reachability, listener config, and security settings.

Summary

• Flink Kafka connectors need a valid Kafka bootstrap.servers string.
• The value should be a comma-separated list of host:port broker endpoints.
• A small getHostnamePort helper is fine when building that string from config.
• Most real failures come from network reachability or Kafka advertised listener issues.
• Validate broker addresses from the same environment where Flink task managers run.