We have solved this in 1.7 by changing the headless service to Type=NodePort and setting the externalTrafficPolicy=Local. This bypasses the internal load balancing of a Service and traffic destined to a specific node on that node port will only work if a Kafka pod is on that node.

apiVersion: v1
kind: Service
metadata:
  name: broker
spec:
  externalTrafficPolicy: Local
  ports:
  - nodePort: 30000
    port: 30000
    protocol: TCP
    targetPort: 9092
  selector:
    app: broker
  type: NodePort

For example, we have two nodes nodeA and nodeB, nodeB is running a kafka pod. nodeA:30000 will not connect but nodeB:30000 will connect to the kafka pod running on nodeB.

https://kubernetes.io/docs/tutorials/services/source-ip/#source-ip-for-services-with-typenodeport

Note this was also available in 1.5 and 1.6 as a beta annotation, more can be found here on feature availability: https://kubernetes.io/docs/tasks/access-application-cluster/create-external-load-balancer/#preserving-the-client-source-ip

Note also that while this ties a kafka pod to a specific external network identity, it does not guarantee that your storage volume will be tied to that network identity. If you are using the VolumeClaimTemplates in a StatefulSet then your volumes are tied to the pod while kafka expects the volume to be tied to the network identity.

For example, if the kafka-0 pod restarts and kafka-0 comes up on nodeC instead of nodeA, kafka-0's pvc (if using VolumeClaimTemplates) has data that it is for nodeA and the broker running on kafka-0 starts rejecting requests thinking that it is nodeA not nodeC.

To fix this, we are looking forward to Local Persistent Volumes but right now we have a single PVC for our kafka StatefulSet and data is stored under $NODENAME on that PVC to tie volume data to a particular node.

https://github.com/kubernetes/features/issues/121 https://kubernetes.io/docs/concepts/storage/volumes/#local

I'd like to say that I'd read this Question and Answer 3 times before from trying to wrap my head around what Headless Services were/what the point of them was. (and I never fully understood Headless Services, or what this Q&A was about.)
And on the 4th read (revisiting it after further educating myself) it finally clicked/I finally understood.

So the purpose of this answer is to restate Nadir's question/problem/and answer as if explaining it to a grade schooler. So that others who stumble upon this will get the significance of Nadir's awesome solution on the first read.

Useful Background Knowledge:

• There exists a Service of type: ExternalName.
  ExternalName Service simply points to a DNS address.
  There are 2 Flavors of ExternalName Service:

  1. Without a Cluster IP:
     A good use case would be allowing a testing cluster and a production cluster to share as much code as possible. (and for simple conviencence in some cases) Pods in both testing and production would point to the same service Inner Cluster DNS Address Name, that would be the predictable reuseable code. The difference would be that the testing enviornment would have a service that points to a SQL service that exists inside the cluster. The production cluster would use an ExternalName Service, which would redirect/point to the DNS Address of a Cloud Providers Managed SQL solution.

  2. With a Cluster IP:
     This is the version of an ExternalName Service that is key to the solution.

• A Stateful Set has 3 parts to its identity:

  1. An Ordinal (Number)
  2. Persistent Storage

  3. A Persistent and predictable Inner Cluster DNS Name (it gets this from the requirement that it must be shipped with a Headless service)

• There are 3 important things to remember about Kube-Proxy:

  1. It makes sure everything has a unique IP.
  2. It's responsible for implementing the Virtual Static Cluster IP's (the Virtual Static Cluster IP's are considered virtual because they only exist in every nodes iptables in the iptables implementation of Kube-Proxy, or in a Kernel Hash Table in the ip-vs next-gen version of Kube-Proxy) and it's also responsible for the Logical Load Balancing effect that occurs with Normal Kubernetes Services that have a Cluster IP.

  3. KubeProxy is responsible for mapping Traffic that comes in on NodePorts to a corresponding Kubernetes Service with a static Cluster IP. <-- This is very important to the requirement that the Stateful Services should be externally exposed, NodePorts are always supposed to be involved when it comes to externally exposing services.

• There are 4 important things to remember about a Headless Service:

  1. It creates a predictable DNS Address.
  2. It doesn't act as an internal cluster Load Balancer. You talk directly to the pod identified by the predictable DNS Address. (which is very desirable for stateful workloads)
  3. It doesn't have a Static Cluster IP.
  4. As a side effect of qualities 2 and 3, it's outside the Realm of Kube-Proxy (which is responsible for directing traffic coming in on Node Ports to Services.) I'll paraphrase this a few times so the problem sinks in: NodePorts can't usually forward traffic to Headless Services. External traffic entering the cluster can't usually be forwarded to Headless Services. It's not intuitive how to externally expose a Headless Service.

Now that we understand the problem better, lets go back to the question: How can a Headless Service (which points to an individual member of a stateful set) be externally exposed?

Solution Part 1:
Any pod in the cluster can talk to the members of the statefulset.
Because the stateful generate a headless service, with a predictable inner cluster DNS address of the form:
statefulsetname-#.associatedheadlessservice.namespace.svc.cluster.local:port
kafka-0.broker.kafka.svc.cluster.local:9092
kafka-1.broker.kafka.svc.cluster.local:9092
kafka-2.broker.kafka.svc.cluster.local:9092
broker.kafka.svc.cluster.local:9092, can also be used to refer to which ever one's available.

Solution Part 2:
You allow external traffic to talk to members of the stateful set, by introducing a 2nd service that can accept external traffic, and then redirecting traffic from that service to the headless service that can only accept internet traffic.
For each pod in the Stateful Set a Service of type ExternalName with a Virtual Static ClusterIP Address that's managed by Kube-Proxy is created. Each one of these ExternalName Services points to/redirects traffic to a predictable static inner cluster DNS Address identified in Solution 1, and because this ExternalName service has a Virtual Static ClusterIP managed via Kube-Proxy, there can be a mapping from NodePorts to it.

Change the service from a headless ClusterIP into a NodePort which would forward request to any of the nodes on a set port (30092 in my example) to port 9042 on the Kafkas. You would hit one of the pods, on random, but I guess that is fine.

20dns.yml becomes (something like this):

# A no longer headless service to create DNS records
---
apiVersion: v1
kind: Service
metadata:
  name: broker
  namespace: kafka
spec:
  type: NodePort
  ports:
  - port: 9092
  - nodePort: 30092
  # [podname].broker.kafka.svc.cluster.local
  selector:
    app: kafka

Disclaimer: You might need two services. One headless for the internal dns names and one NodePort for the external access. I haven't tried this my self.

From the kubernetes kafka documentation:

Outside access with hostport

An alternative is to use the hostport for the outside access. When using this only one kafka broker can run on each host, which is a good idea anyway.

In order to switch to hostport the kafka advertise address needs to be switched to the ExternalIP or ExternalDNS name of the node running the broker. in kafka/10broker-config.yml switch to

OUTSIDE_HOST=$(kubectl get node "$NODE_NAME" -o jsonpath='{.status.addresses[?(@.type=="ExternalIP")].address}')
OUTSIDE_PORT=${OutsidePort}

and in kafka/50kafka.yml add the hostport:

    - name: outside
      containerPort: 9094
      hostPort: 9094