title: How to extract timestamp embedded in messages in Kafka Streams
body: <p>I want to extract Timestamps embedded with each message and send them as json payload into my database.</p>
  
  <p>I want to get the following three timestamps.</p>
  
  <p><strong>Event-time:</strong> <code>The point in time when an event or data record occurred, i.e. was originally created “by the source”.</code></p>
  
  <p><strong>Processing-time:</strong> <code>The point in time when the event or data record happens to be processed by the stream processing application, i.e. when the record is being consumed.</code></p>
  
  <p><strong>Ingestion-time:</strong> <code>The point in time when an event or data record is stored in a topic partition by a Kafka broker.</code></p>
  
  <p>This is my streams application code:</p>
  
  <pre><code>Properties props = new Properties();
  
  props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-pipe");
  props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_URL + ":9092"); // pass from env localhost:9092 ,BROKER_URL + ":9092"
  props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  
  final StreamsBuilder builder = new StreamsBuilder();
  
  KStream&lt;String, String&gt; source_o365_user_activity = builder.stream("o365_user_activity");
  
  source_o365_user_activity.flatMapValues(new ValueMapper&lt;String, Iterable&lt;String&gt;&gt;() {
      @Override
      public Iterable&lt;String&gt; apply(String value) {
          System.out.println("========&gt; o365_user_activity_by_date Log:     " + value);
          ArrayList&lt;String&gt; keywords = new ArrayList&lt;String&gt;();
          try {
              JSONObject send = new JSONObject();
              JSONObject received = new JSONObject(value);
  
              send.put("current_date", getCurrentDate().toString()); // UTC TIME
              send.put("activity_time", received.get("CreationTime")); // CONSTANTS FINAL STATIC(Topic Names, Cassandra keys)
              send.put("user_id", received.get("UserId"));
              send.put("operation", received.get("Operation"));
              send.put("workload", received.get("Workload"));
              keywords.add(send.toString());
  
          } catch (Exception e) {
              // TODO: handle exception
              System.err.println("Unable to convert to json");
              e.printStackTrace();
          }
  
          return keywords;
      }
  }).to("o365_user_activity_by_date");
  </code></pre>
  
  <p>In the code I am simply getting each record, doing some stream processing on it and sending it to a different topic.</p>
  
  <p>Now with each record I want to send <code>Event-time</code>, <code>Processing-time</code> and <code>Ingestion-time</code> embedded in the payload.</p>
  
  <p>I have looked at the <code>FailOnInvalidTimestamp</code> and <code>WallclockTimestampExtractor</code> but I am confused on how to use them.</p>
  
  <p>Kindly guide me how can I achieve this.</p>