<p>I have one application which servers for REST request and also is listening on a Kafka topic. 
I deployed the application to Kubernetes and configure the readiness probe like this</p>

<pre><code>readinessProbe:
  exec:
    command:
    - cat
    - /tmp/healthy
  initialDelaySeconds: 5
  periodSeconds: 5
</code></pre>

<p>basically following the instruction from <a href="https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/" rel="nofollow noreferrer">[configure-liveness-readiness-startup-probes]</a></p>

<p>After deployment is done, I can see the pod readiness probe fails</p>

<pre><code>Readiness probe failed: cat: can't open '/tmp/healthy': No such file or directory
</code></pre>

<p>That is expected. Then I sent a kafka message to the topic . I observed that </p>

<p>1) the kafka message has been consumed by my application and saved to database.<br>
2) the rest api can't be accessed.</p>

<p>I assumed if the pod's readiness probe is failed, the application can neither receive kafka message nor the rest request. But why in my test, the REST request and Kafka message are handled differently. </p>

<p>According to the Kubernete documentation:</p>

<pre><code>The kubelet uses readiness probes to know when a Container is ready to start accepting traffic
</code></pre>

<p>But it doesn't say clearly what kind of traffic it really means.
Does kubernetes only restrict the http traffic to the pod if readiness probe failes but not restrict tcp traffic (as Kafka is working over tcp)? </p>

<p>My actual intention is to make my service application (kafka consumer) able to control when to receive kafka messages (and REST request as well). E.g. if there is heavy opertion, my service will delete the /tmp/healthy file and thus make the pod not ready for recieving kafka message and Rest request.  When the heavy operation is finished, the app write the healthy file to make the pod ready for receiving message.</p>

<p>Some more information, in my test, the kubernetes version is v1.14.3 and kafka broker is running in a separated vm outside of kubernetes.</p>

<p>This is two very different things:</p>

<ul>
<li><strong>Receiving requests</strong>: An <em>external service</em> is sending a request and expect a response.</li>
<li><strong>Sending requests</strong>: Your service is sending a request and waiting for a response.</li>
</ul>

<h2>ReadinessProbe</h2>

<p>When a ReadinessProbe fails, <strong>no new requests will be routed to the pod</strong>.</p>

<h2>Kafka consumer</h2>

<p>If your pod is a <em>Kafka consumer</em>, then your <strong>pod is initializing requests</strong> to Kafka, to retrieve messages from the <em>topic</em>.</p>

<p><strong>Check for required directory</strong></p>

<blockquote>
  <p>can't open '/tmp/healthy': No such file or directory</p>
</blockquote>

<p>If the directory <code>/tmp/healthy</code> is needed for your service to work correctly, your service should check for it on startup, and <code>exit(1)</code> (crash with an error message)  if the required directory isn't available. This should be done before connecting to Kafka. If your application uses the directory continually, e.g. writing to it, any operations <strong>error codes should be checked and handled properly</strong> - log and crash depending on your situation.</p>

<h2>Consuming Kafka messages</h2>

<blockquote>
  <p>My actual intention is to make my service application (kafka consumer) able to control when to receive kafka messages ( and REST request as well). E.g. if there is heavy opertion, my service will delete the /tmp/healthy file and thus make the pod not ready for recieving kafka message and Rest request.</p>
</blockquote>

<p>Kafka consumers <strong>poll</strong> Kafka for more data, whenever the consumer want. In other words, the Kafka consumer <em>ask</em> for more data whenever it is ready for more data.</p>

<p>Example consumer code:</p>

<pre><code> while (true) {
     ConsumerRecords&lt;String, String&gt; records = consumer.poll(100);
     for (ConsumerRecord&lt;String, String&gt; record : records) {
         // process your records
     }
 }
</code></pre>

<p>Remember to <code>commit</code> the records that you have <em>processed</em> so that the messages aren't processed multiple times e.g. after a crash.</p>
