title: custom timestamp extractor in kafka streams
body: <p>I'm trying to use kafka streams to process some data from a kafka topic. The data comes from kafka topic that was written to by kafka 0.11.0 something which doesn't have the embedded timestamp. After some reading on the internet, I came to understand that I can solve this problem by extending <code>TimestampExtractor</code> class in a custom class and passing it in the <code>StreamsConfig</code>.</p>
  
  <p>I did so like this --</p>
  
  <pre><code>class MyEventTimestampExtractor extends TimestampExtractor {
    override def extract(record: ConsumerRecord[AnyRef, AnyRef], prev: Long) = {
      record.value() match {
          case w: String =&gt; 1000L
          case _ =&gt; throw new RuntimeException(s"Called for $record")
      }
    }
  }
  </code></pre>
  
  <p>I based it off of <a href="https://github.com/softwaremill/its-pres/blob/master/kafka/src/main/scala/com/softwaremill/its/KafkaHotspotDetector.scala#L263-L277" rel="nofollow noreferrer" title="this code on github">this code on github</a></p>
  
  <p>But, I get this error when I do an <code>sbt run</code></p>
  
  <pre><code>[error] /home/someuser/app/blahblah/src/main/scala/main.scala:34: class MyEventTimestampExtractor needs to be abstract, since method extract in trait TimestampExtractor of type (x$1: org.apache.kafka.clients.consumer.ConsumerRecord[Object,Object], x$2: Long)Long is not defined
  [error] (Note that Long does not match Long)
  [error] class MyEventTimestampExtractor extends TimestampExtractor {
  [error]       ^
  [error] /home/someuser/app/blahblah/src/main/scala/main.scala:35: method extract overrides nothing.
  [error] Note: the super classes of class MyEventTimestampExtractor contain the following, non final members named extract:
  [error] def extract(x$1: org.apache.kafka.clients.consumer.ConsumerRecord[Object,Object],x$2: Long): Long
  [error]   override def extract(record: ConsumerRecord[AnyRef, AnyRef], prev: Long): Long = {
  [error]                ^
  [error] two errors found
  [error] (compile:compileIncremental) Compilation failed
  </code></pre>
  
  <p>My build.sbt file is this --</p>
  
  <pre><code>name := "kafka streams experiment"
  version := "1.0"
  scalaVersion := "2.12.4"
  
  libraryDependencies ++= Seq(
    "org.apache.kafka" % "kafka-streams" % "1.0.0"
  )
  </code></pre>
  
  <p>I don't really understand the error. Specifically the part around <code>Note that Long does not match Long</code>. What could I be doing wrong? thanks!</p>