# Ingesting latest logs from kafka with logstash

**URL:** <https://discuss.elastic.co/t/ingesting-latest-logs-from-kafka-with-logstash/215209>\
**Category:** Logstash\
**Created:** [January 15, 2020, 8:54pm UTC](https://discuss.elastic.co/t/ingesting-latest-logs-from-kafka-with-logstash/215209 "2020-01-15T20:54:52Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![canaria](https://avatars.discourse-cdn.com/v4/letter/c/977dab/32.png) [@canaria](https://discuss.elastic.co/u/canaria)\
**Post date:** [January 15, 2020, 8:54pm UTC](https://discuss.elastic.co/t/ingesting-latest-logs-from-kafka-with-logstash/215209/1 "2020-01-15T20:54:52Z")

</div>

Hi,

I searched previous opened topics related that question but none of them have any reply. I’m ingesting logs from apache kafka with logstash, everything is fine but I noticed that whenever I started logstash instance it doesn't starts to ingest latest logs. The logs’ timestamps belong 13-14 hours ago.

I was thinking that after I set **auto\_offset\_reset =\> “latest”** parameter should have solved my issue but I guess It didn’t work.

This is input part of my logstash config:

```
input {
  kafka {
    bootstrap_servers => ["bootsrap1:39092,bootsrap2:39092,bootsrap3:39092,bootsrap4:39092,bootsrap5:39092"]
    topics => "th-cef"
    auto_offset_reset => "latest"
    client_id => "test10"
    type => "logs9"
    group_id => "sectechlogstash"
    consumer_threads => 19
  }
}

```

Any tips or help would be appreciated  
Thanks

---

<div class="post-metadata">

**Author:** ![canaria](https://avatars.discourse-cdn.com/v4/letter/c/977dab/32.png) [@canaria](https://discuss.elastic.co/u/canaria)\
**Post date:** [January 15, 2020, 9:40pm UTC](https://discuss.elastic.co/t/ingesting-latest-logs-from-kafka-with-logstash/215209/2 "2020-01-15T21:40:12Z")

</div>

It's funny, I didn't change anything at all. After I created this topic, I just started again logstash instances and noticed that logstash ingest latest logs from kafka.

Thanks anyway to Elastic Community!

---

<div class="post-metadata">

**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\
**Post date:** [January 15, 2020, 10:20pm UTC](https://discuss.elastic.co/t/ingesting-latest-logs-from-kafka-with-logstash/215209/3 "2020-01-15T22:20:46Z")

</div>

auto\_offset\_reset determines what it does if there is no offset for the consumer group, or if the offset is beyond the start of the queue in kafka. It is possible that the 14 hour old data was very close to expiring, and when you stopped and restarted logstash the offset for sectechlogstash moved beyond the expiry window.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [February 12, 2020, 10:24pm UTC](https://discuss.elastic.co/t/ingesting-latest-logs-from-kafka-with-logstash/215209/4 "2020-02-12T22:24:36Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.