<p>I have an application that runs on multiple pods and scales out when traffic increases. One of the features of the application is that &quot;It picks messages from Kafka and triggers email&quot;.</p>
<p>When multiple pods are running, all of them trigger the email as all pods are supposed to pick up the msg as per the design.</p>
<p>How can I restrict the email functionality to work on any one of the pods at a time?</p>
<p>Cluster - EKS , Programming language - Scala AKKA</p>

Pranav
<blockquote>
<p>How can I restrict the email functionality to work on any one of the
pods at a time?</p>
</blockquote>
<p><strong>In short:</strong> Use same consumer group for all the pods that trigger the email.
Usually, workloads are categorized into groups based on the work they do. Members of the same group share workload amongst themselves.</p>
<p>You could certainly have given Kafka consumer configurations like <code>bootstrap.servers</code> etc to your pods. In that configuration, give a property with name <code>group.id</code> to some value like <code>email-trigger-group</code> for example and then the workload will be shared as you would expect.</p>
<p>You could have used <em>labels</em> for your pods that trigger email. You can use the same label value for your consumer <code>group.id</code> for all of your pods.</p>
<hr />
<p>We can divide the problem into two sub-problems:</p>
<p><strong>1. Trigger email</strong></p>
<p>This workload can be shared by multiple consumers in the group.</p>
<p><strong>2. Answer requests to the frontend</strong></p>
<p>Use manual <code>consumer.assign()</code> for the whole topic (all partitions).</p>
<p>The frontend would specify the <em>timestamp</em> from where it would want new messages from i.e. messages with timestamps <code>&gt;</code> this timestamp will be retrieved from <em>all partitions</em> of the topic. To do this use <code>consumer.offsetsForTimes()</code> to get the timestamps, poll and send the messages as response.</p>
<pre class="lang-java prettyprint-override"><code>List&lt;TopicPartition&gt; topicPartitions = consumer.partitionsFor(&quot;your_topic&quot;).stream().map(partitionInfo -&gt; new TopicPartition(partitionInfo.topic(), partitionInfo.partition()).toList();
consumer.assign(topicPartitions);

// Populate the map
Map&lt;TopicPartition, Long&gt; partitionTimestamp = new LinkedHashMap&lt;&gt;();

// Add the same timestamp received from frontend for all partitions
topicPartitions.forEach(topicPartition -&gt; partitionTimestamp.put(topicPartition, timestampFromFrontend));

// Get the offsets and seek
Map&lt;TopicPartition,OffsetAndTimestamp&gt; offsetsForTimes = consumer.offsetsForTimes(offsetsForTimestamp);

// Seek to the offsets
offsetsForTimes.forEach( (tp, oft) -&gt; consumer.seek(tp, oft.offset()) );

// Poll and return
consumer.poll();
</code></pre>
