# Restore default timeExtractor before sink

**URL:** <https://forum.confluent.io/t/restore-default-timeextractor-before-sink/3715>\
**Category:** Kafka Streams\
**Created:** [21 December 2021 17:44 UTC](https://forum.confluent.io/t/restore-default-timeextractor-before-sink/3715 "2021-12-21T17:44:16Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![programista4k](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/programista4k/32/767_2.png) [@programista4k](https://forum.confluent.io/u/programista4k)\
**Post date:** [21 December 2021 17:44 UTC](https://forum.confluent.io/t/restore-default-timeextractor-before-sink/3715/1 "2021-12-21T17:44:16Z")

</div>

Halu frens,

I’m doing some windowed aggregations with Kafka Streams. To use event time as a timestamp used in the windowing I have set the `TimestampExtractor` on the source:

```auto
streamBuilder.stream("myTopic", Consumed.with(keySerde, valueSerde)
                .withTimestampExtractor(myTimestampExtractor))

```

now after all calculation and before sink I want to restore the default TimestampExtractor which is ingesting time called `LogAppendTime` because I want Kafka to save output sink with the ingesting timestamp but windowing want to do on the event time.  
How to do that?

Or maybe there is a possibility to set the timestampExtractor only for windowing?

```auto
myStream.groupByKey()
                .windowedBy(TimeWindows.of(duration))
                .aggregate(
                        () -> null,
                        new MyAggregator(duration.toString()),
                        Materialized.with(keySerde, valueSerde));

```

![14e20a19e38b32d6c3928f97e4cc9ec3](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/2X/6/6c9a1b4aae6294e1a5284fececde245790af2e52.jpeg)

---

<div class="post-metadata">

**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\
**Post date:** [21 December 2021 19:19 UTC](https://forum.confluent.io/t/restore-default-timeextractor-before-sink/3715/2 "2021-12-21T19:19:05Z")

</div>

There is no `TimestampExtractor` concept on the write path, only on the read path.

`LogAppendTime` is a broker/topic config. You can configure it for the output topic broker side. For this case, the broker would overwrite the timestamp provided by the producer (aka Kafka Streams) by it’s current wall-clock time when it receives the data.

If you want to change the timestamp within KafkaStreams, you could use a `transform()` and call `context.forward(..., To.all().withTimestamp(...))` before calling `to()`.

---

<div class="post-metadata">

**Author:** ![programista4k](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/programista4k/32/767_2.png) [@programista4k](https://forum.confluent.io/u/programista4k)\
**Post date:** [21 December 2021 23:46 UTC](https://forum.confluent.io/t/restore-default-timeextractor-before-sink/3715/3 "2021-12-21T23:46:06Z")

</div>

Thank you Sir, and is there any other solution then setting `TimestampExtractor` if I want to use the event time (from the record’s value not from record’s timestamp) in windowed aggregations?

Want something like this to run before windowed aggregations:

```auto
public class HeadersTransformer implements Transformer<String, MyEvent, KeyValue<String, MyEvent>> {

...
    @Override
    public KeyValue<String, MyEvent> transform(String k, MyEvent v) {
        context.headers()
                .remove(KafkaHeaders.LAST_TIMESTAMP)
                .add(KafkaHeaders.LAST_TIMESTAMP, context.timestamp().toString().getBytes());

        context.setTimestamp(v.getEventTime()); // this is the syntax I miss

        return KeyValue.pair(k, v);
    }
...

```

 ![EYgc6qNUEAA6NsN](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/2X/0/0ec0a796e5c78464f69abb1a7c0d3fe8dc0e9286.png)

---

<div class="post-metadata">

**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\
**Post date:** [22 December 2021 01:28 UTC](https://forum.confluent.io/t/restore-default-timeextractor-before-sink/3715/4 "2021-12-22T01:28:09Z")

</div>

What you put down works basically, but you need to write it differently:

```java
transform(...) {
  // there is no context.setTimestamp(...)
  // instead use
  context.forward(..., To.all().withTimestamp(...));

  // use a "dummy" return to not forward more records (cf. JavaDocs)
  return null; // `null` is valid and means no additional output
}

```

In Kafka Streams, each record has a timestamp associated that is used by any temporal DSL operation (e.g., windowed aggregation or stream-stream joins, or others). By default the **record** timestamp is set to the **message** timestamp, but you can also set it to a different value using a custom `TimestampExtractor`. Similarly, you can set a new record timestamp at any point in your topology via `context.forward(..., To.all().withTimestamp(...))` as shown above.