# Filter by timestamp - re-ingesting logs from Kafka

**URL:** <https://discuss.elastic.co/t/filter-by-timestamp-re-ingesting-logs-from-kafka/169844>\
**Category:** Logstash\
**Created:** [February 25, 2019, 1:53pm UTC](https://discuss.elastic.co/t/filter-by-timestamp-re-ingesting-logs-from-kafka/169844 "2019-02-25T13:53:03Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![carrot](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/carrot/32/44335_2.png) [@carrot](https://discuss.elastic.co/u/carrot)\
**Post date:** [February 25, 2019, 1:53pm UTC](https://discuss.elastic.co/t/filter-by-timestamp-re-ingesting-logs-from-kafka/169844/1 "2019-02-25T13:53:03Z")

</div>

We had a problem with our cluster this weekend (index went read-only) and looks like the OFFSET expired in Kafka or something similar. The following message appeared in our logstash logs regarding Kafka`INVALID_FETCH_SESSION_EPOCH`.

Saying that, we lost 24 hours of data and it is still in our Kafka cluster. Using the offset is somewhat hard but Kafka does have the timestamp recorded in the message.

I would like to restart at the `earliest` and then filter on data greater than the last message I received. How would the filter looks like in Logstash?

Example:  
if @timestamp \< "2019-02-24T10:00:00.000Z" { drop }

Thanks

---

<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:** [February 25, 2019, 2:53pm UTC](https://discuss.elastic.co/t/filter-by-timestamp-re-ingesting-logs-from-kafka/169844/2 "2019-02-25T14:53:36Z")

</div>

```
    mutate { add_field => { "[@metadata][cutoff]" => "2019-02-24T10:00:00.000Z" } }
    date { match => ["[@metadata][cutoff]", "ISO8601" ] target => "[@metadata][cutoff]" }
    ruby { code => 'if event.get("@timestamp") < event.get("[@metadata][cutoff]") then event.cancel end ' }
```

---

<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:** [March 25, 2019, 2:53pm UTC](https://discuss.elastic.co/t/filter-by-timestamp-re-ingesting-logs-from-kafka/169844/3 "2019-03-25T14:53:40Z")

</div>

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