<p>I have been trying to deploy Kafka using <a href="https://docs.confluent.io/current/installation/installing_cp/cp-helm-charts/docs/index.html" rel="nofollow noreferrer">Helm charts</a>. So I defined NodePort service for Kafka pods. I checked console Kafka producer and consumer with the same hosts and ports - they work properly. However, when I create Spark application as data consumer and Kafka as producer they are not able to connect to the Kafka service0. I used minikube ip (instead of node ip) for the host and service NodePort port.
Although, in Spark logs, I saw that NodePort service resolves endpoints and brokers are discovered as pods addressed and ports:</p>

<pre><code>INFO AbstractCoordinator: [Consumer clientId=consumer-1, groupId=avro_data] Discovered group coordinator 172.17.0.20:9092 (id: 2147483645 rack: null)
INFO ConsumerCoordinator: [Consumer clientId=consumer-1, groupId=avro_data] Revoking previously assigned partitions []
INFO AbstractCoordinator: [Consumer clientId=consumer-1, groupId=avro_data] (Re-)joining group
WARN NetworkClient: [Consumer clientId=consumer-1, groupId=avro_data] Connection to node 2147483645 (/172.17.0.20:9092) could not be established. Broker may not be available.
INFO AbstractCoordinator: [Consumer clientId=consumer-1, groupId=avro_data] Group coordinator 172.17.0.20:9092 (id: 2147483645 rack: null) is unavailable or invalid, will attempt rediscovery
WARN NetworkClient: [Consumer clientId=consumer-1, groupId=avro_data] Connection to node 2 (/172.17.0.20:9092) could not be established. Broker may not be available.
WARN NetworkClient: [Consumer clientId=consumer-1, groupId=avro_data] Connection to node 0 (/172.17.0.12:9092) could not be established. Broker may not be available.
</code></pre>

<p>How this behavior can be changed?</p>

<p>NodePort service definition looks like this:</p>

<pre><code>kind: Service
apiVersion: v1
metadata:
  name: kafka-service
spec:
  selector:
    app: cp-kafka
    release: my-confluent-oss
  ports:
    - protocol: TCP
      targetPort: 9092
      port: 32400
      nodePort: 32400
  type: NodePort
</code></pre>

<p>Spark consumer configuration:</p>

<pre><code>def kafkaParams() = Map[String, Object](
  "bootstrap.servers" -&gt; "192.168.99.100:32400",
  "schema.registry.url" -&gt; "http://192.168.99.100:8081",
  "key.deserializer" -&gt; classOf[StringDeserializer],
  "value.deserializer" -&gt; classOf[KafkaAvroDeserializer],
  "group.id" -&gt; "avro_data",
  "auto.offset.reset" -&gt; "earliest",
  "enable.auto.commit" -&gt; (false: java.lang.Boolean)
)
</code></pre>

<p>Kafka producer configuration:</p>

<pre><code>  props.put("bootstrap.servers", "192.168.99.100:32400")
  props.put("client.id", "avro_data")
  props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
  props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer")
  props.put("schema.registry.url", "http://192.168.99.100:32500")
</code></pre>

<p>All the K8s services for Kafka:</p>

<pre><code>NAME                                     TYPE           CLUSTER-IP       EXTERNAL-IP   PORT(S)             AGE
kafka-service                            NodePort       10.99.113.234    &lt;none&gt;        32400:32400/TCP     6m34s
kubernetes                               ClusterIP      10.96.0.1        &lt;none&gt;        443/TCP             27d
my-confluent-oss-cp-kafka                ClusterIP      10.100.156.108   &lt;none&gt;        9092/TCP            102m
my-confluent-oss-cp-kafka-connect        ClusterIP      10.99.78.89      &lt;none&gt;        8083/TCP            102m
my-confluent-oss-cp-kafka-headless       ClusterIP      None             &lt;none&gt;        9092/TCP            102m
my-confluent-oss-cp-kafka-rest           ClusterIP      10.100.152.109   &lt;none&gt;        8082/TCP            102m
my-confluent-oss-cp-ksql-server          ClusterIP      10.96.249.202    &lt;none&gt;        8088/TCP            102m
my-confluent-oss-cp-schema-registry      ClusterIP      10.109.27.45     &lt;none&gt;        8081/TCP            102m
my-confluent-oss-cp-zookeeper            ClusterIP      10.102.182.90    &lt;none&gt;        2181/TCP            102m
my-confluent-oss-cp-zookeeper-headless   ClusterIP      None             &lt;none&gt;        2888/TCP,3888/TCP   102m
schema-registry-service                  NodePort       10.103.100.64    &lt;none&gt;        32500:32500/TCP     33m
zookeeper-np                             NodePort       10.98.180.130    &lt;none&gt;        32181:32181/TCP     53m
</code></pre>

<p>I had a similar issue when I was trying to access kafka broker (<a href="https://github.com/confluentinc/cp-helm-charts" rel="nofollow noreferrer">cp-helm-chart</a>) running on minikube from outside. </p>

<p>Here how I resolved it. Before you install using helm install from local repository.</p>

<ol>
<li>Edit inside this file <a href="https://github.com/confluentinc/cp-helm-charts/blob/master/charts/cp-kafka/values.yaml" rel="nofollow noreferrer">https://github.com/confluentinc/cp-helm-charts/blob/master/charts/cp-kafka/values.yaml</a></li>
<li>Search for nodeport: and change its enabled field to true.
<br/>
nodeport:<br/>
enabled: true</li>
<li>Uncomment these two lines by removing #:<br/>
"advertised.listeners": |-<br/>
EXTERNAL://${HOST_IP}:$((31090 + ${KAFKA_BROKER_ID}))</li>
<li>Replace ${HOST_IP} by your minikube ip (enter minikube ip in cmd to retrieve your k8s host ip e.g : 196.169.99.100)</li>
<li>Replace ${KAFKA_BROKER_ID} by the broker id (if only one broker is running than its will be by default just 0)</li>
<li>Finally it would look something like this:<br/>
"advertised.listeners": |-<br/>
EXTERNAL://196.169.99.100:31090</li>
</ol>

<p>Now you can access the kafka broker running within k8s cluster from outside by pointing the bootstrap.servers to 196.169.99.100:31090</p>
