<p>I run 1 kafka and 3 zookeeper-server in docker on kubernetes following this <a href="http://www.defuze.org/archives/351-running-a-zookeeper-and-kafka-cluster-with-kubernetes-on-aws.html" rel="noreferrer">instruction</a>. I cannot produce/consume topics outside pod(docker container).</p>

<pre><code>bin/kafka-console-producer.sh --broker-list 1.2.3.4:9092 --topic

[2016-06-11 15:14:46,889] ERROR Error when sending message to topic test with key: null, value: 3 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.TimeoutException: Batch containing 3 record(s) expired due to timeout while requesting metadata from brokers for test-0

bin/kafka-console-consumer.sh --zookeeper 5.6.7.8:2181 --topic test --from-beginning 

[2016-06-11 15:15:58,985] WARN Fetching topic metadata with correlation id 0 for topics [Set(test)] from broker [BrokerEndPoint(1001,kafka-service,9092)] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:124)
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
    at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
    at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
[2016-06-11 15:15:58,992] WARN [console-consumer-66869_tattoo-NV49C-1465629357799-ce1529da-leader-finder-thread], Failed to find leader for Set([test,0]) (kafka.consumer.ConsumerFetcherManager$LeaderFinderThread)
kafka.common.KafkaException: fetching topic metadata for topics [Set(test)] from broker [ArrayBuffer(BrokerEndPoint(1001,kafka-service,9092))] failed
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:73)
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
    at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
    at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
Caused by: java.nio.channels.ClosedChannelException
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:124)
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
    ... 3 more


kafka log:
    [2016-06-11 07:47:58,269] INFO [Kafka Server 1001], started (kafka.server.KafkaServer)
[2016-06-11 07:53:50,404] INFO [ReplicaFetcherManager on broker 1001] Removed fetcher for partitions [test,0] (kafka.    server.ReplicaFetcherManager)
[2016-06-11 07:53:50,443] INFO Completed load of log test-0 with log end offset 0 (kafka.log.Log)
[2016-06-11 07:53:50,458] INFO Created log for partition [test,0] in /kafka/kafka-logs-kafka-controller-3rsv3 with     properties {compression.type -&gt; producer, message.format.version -&gt; 0.10.0-IV1, file.delete.delay.ms -&gt; 60000, max.    message.bytes -&gt; 1000012, message.timestamp.type -&gt; CreateTime, min.insync.replicas -&gt; 1, segment.jitter.ms -&gt; 0,     preallocate -&gt; false, min.cleanable.dirty.ratio -&gt; 0.5, index.interval.bytes -&gt; 4096, unclean.leader.election.enable     -&gt; true, retention.bytes -&gt; -1, delete.retention.ms -&gt; 86400000, cleanup.policy -&gt; delete, flush.ms -&gt;     9223372036854775807, segment.ms -&gt; 604800000, segment.bytes -&gt; 1073741824, retention.ms -&gt; 604800000, message.    timestamp.difference.max.ms -&gt; 9223372036854775807, segment.index.bytes -&gt; 10485760, flush.messages -&gt;     9223372036854775807}. (kafka.log.LogManager)
[2016-06-11 07:53:50,459] INFO Partition [test,0] on broker 1001: No checkpointed highwatermark is found for     partition [test,0] (kafka.cluster.Partition)
[2016-06-11 07:57:57,955] INFO [Group Metadata Manager on Broker 1001]: Removed 0 expired offsets in 0 milliseconds. (    kafka.coordinator.GroupMetadataManager)
</code></pre>

<p>And the <code>config/server.properties</code></p>

<pre><code>broker.id=-1
log.dirs=/kafka/kafka-logs-kafka-controller-3rsv3
num.partitions=1
zookeeper.connect=zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
zookeeper.connection.timeout.ms=6000

service.port.9092.tcp.addr=10.254.68.65
service.port.9092.tcp.proto=tcp
service.service.port.kafka.port=9092
service.service.port=9092
service.port=tcp://10.254.68.65:9092
service.port.9092.tcp.port=9092
version=0.10.0.0
service.service.host=10.254.68.65
port=9092
advertised.host.name=kafka-service
service.port.9092.tcp=tcp://10.254.68.65:9092
advertised.port=9092
</code></pre>

<p>But I can do <code>bin/kafka-console-producer.sh --broker-list localhost:9092 --topic</code> and <code>bin/kafka-console-consumer.sh --zookeeper 5.6.7.8:2181 --topic test --from-beginning</code> if I am inside the pod (docker container).</p>

<p>And I can create and list topics normally when connecting to zookeeper's service:</p>

<pre><code>bin/kafka-topics.sh --describe --zookeeper 5.6.7.8:2181 --topic test
Topic:test  PartitionCount:1    ReplicationFactor:1 Configs:
    Topic: test Partition: 0    Leader: 1001    Replicas: 1001  Isr: 1001
</code></pre>

<p>And my yaml file for creating kafka replicaiton-congtroller and service: </p>

<pre><code>---
apiVersion: v1
kind: Service
metadata:
  name: kafka-service2
  labels:
    app: kafka2
spec:
  clusterIP: None
  ports:
  - port: 9092
    name: kafka-port
    targetPort: 9092
    protocol: TCP
  selector:
    app: kafka2
---
apiVersion: v1
kind: ReplicationController
metadata:
  name: kafka-controller2
spec:
  replicas: 1
  selector:
    app: kafka2
  template:
    metadata:
      labels:
        app: kafka2
    spec:
      containers:
      - name: kafka2
        image: wurstmeister/kafka
        ports:
        - containerPort: 9092
        env:
        - name: KAFKA_ADVERTISED_PORT
          value: "9092"
        - name: KAFKA_ADVERTISED_HOST_NAME
          value: kafka-service2
        - name: KAFKA_ZOOKEEPER_CONNECT
          value: zoo1:2181,zoo2:2181,zoo3:2181
</code></pre>

<p>Kafka register to zookeeper with its service's name. And consuming/producing messages need access to the service names(here is dns records on zookeeper-1, zookeeper-2, zookeeper-3), which are only accessible through kubernetes' dns. So only application running on kubernetes can access my kafka. Therefore I cannot use the external IP of kafka-service or port-forwarding the kafka pod to localhost and then access it. </p>

<p>But why can I create, list and describe topics outside kubernetes cluster? I guess it is because zookeepers can do those operation by themselves. But consuming/producing message will need access to the ADVERTISED_HOST_NAME provided by kafka.</p>
