title: Kafka - Simplest Way to Get Latest Offset
body: <p>I'm building an application that allows subscriptions to kafka topics to be added and removed dynamically. When a topic subscription is added I wanted to run a batch job every hour that gets all of the new messages and pushes them into another datastore.</p>
  
  <p>What I want to understand is how to get the current offset of a topic. As soon as a subscription is added, I want the next batch job to get all messages since the approximate time of the subscription.</p>
  
  <p>As an example, imagine I have a topic called "TopicA" which is constantly receiving messages. If I add a subscription at 7.15pm, when the batch job runs at 8pm I want all messages since 7.15pm to be batched up. I'm happy for the time to be approximate - 7.10, 7.20 etc. 5 or 10 minutes either side causes me no concern.</p>
  
  <p>So my intended solution is to get the current offset of a topic the moment a subscription is added. I've looked at the simple consumer, but I don't want to get involved in all of the cluster managemnet aspects for this basic use case.</p>
  
  <p>I've also looked at the high-level consumer. I could something like this:</p>
  
  <pre><code>consumer.createMessageStreamsByFilter(new Whitelist(topicName)).head.head.offset
  </code></pre>
  
  <p>What concerns me with this approach is the call to "head" is actually a stream. So I believe it will block waiting for the next message. Blocking is problematic because it may cause other subscriptions to be queued up until the next message arrives.</p>
  
  <p>I'm happy to spend some time implementing the latter approach, but if there is a simpler way that doesn't require me to write error-prone concurrent code, then I'd rather not waste my time.</p>
  
  <p>I'll also need a way to get all logs since that offset.</p>