title: Kafka Streams Windowing with Custom TimestampExtractor
body: <p>I'm trying to create a Kafka Streams application where I'm trying to calculate unique devices per platform within a time window.</p>
  
  <p><strong>Event Class</strong></p>
  
  <pre><code>public class Event {
      private String eventId;
      private String deviceId;
      private String platform;
      private ZonedDateTime createdAt;
  }
  </code></pre>
  
  <p>I need the time window respect the event's createdAt so I wrote a <code>TimestampExtractor</code> implementation like below:</p>
  
  <pre><code>public class EventTimestampExtractor implements TimestampExtractor {
      @Override
      public long extract(final ConsumerRecord&lt;Object, Object&gt; record, final long previousTimestamp) {
          final Event event = (Event) record.value();
          final ZonedDateTime eventCreationTime = event.getCreatedAt();
          final long timestamp = eventCreationTime.toEpochSecond();
  
          log.trace("Event ({}) yielded timestamp: {}", event.getEventId(), timestamp);
  
          return timestamp;
      }
  }
  </code></pre>
  
  <p>Lastly, here's my streaming app code:</p>
  
  <pre><code>final KStream&lt;String, Event&gt; eventStream = builder.stream("events_ingestion");
  
  eventStream
      .selectKey((key, event) -&gt; {
          final String platform = event.getPlatform();
          final String deviceId = event.getDeviceId());
  
          return String.join("::", platform, deviceId);
      })
      .groupByKey()
      .windowedBy(TimeWindows.of(TimeUnit.MINUTES.toMillis(15)))
      .count(Materialized.as(COUNT_STORE));
  </code></pre>
  
  <p>When I push an event into <code>event_ingestion</code> topic, I can see that the timestamp is logged into the application logs and data is being written into the count store.</p>
  
  <p>When I iterate over the count store, I see the following:</p>
  
  <pre><code>Key: [ANDROID::1@1539000000/1539900000], Value: 2
  </code></pre>
  
  <p>Although my time window is 15 minutes, the key spans 10 days. If I remove my TimestampExtractor implementation from stream config (hence go back to processing time) the key spans 15 minutes as expected:</p>
  
  <pre><code>Key: [ANDROID::1@1539256500000/1539257400000], Value: 1
  </code></pre>
  
  <p>What am I doing wrong here? Any ideas?</p>