/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.kafka.streams.processor;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.errors.StreamsException;

/**
 * Retrieves embedded metadata timestamps from Kafka messages.
 * Embedded metadata timestamp was introduced in "KIP-32: Add timestamps to Kafka message" for the new
 * 0.10+ Kafka message format.
 * <p>
 * Here, "embedded metadata" refers to the fact that compatible Kafka producer clients automatically and
 * transparently embed such timestamps into message metadata they send to Kafka, which can then be retrieved
 * via this timestamp extractor.
 * <p>
 * If the embedded metadata timestamp represents <i>CreateTime</i> (cf. Kafka broker setting
 * {@code message.timestamp.type} and Kafka topic setting {@code log.message.timestamp.type}),
 * this extractor effectively provides <i>event-time</i> semantics.
 * If <i>LogAppendTime</i> is used as broker/topic setting to define the embedded metadata timestamps,
 * using this extractor effectively provides <i>ingestion-time</i> semantics.
 * <p>
 * If you need <i>processing-time</i> semantics, use {@link WallclockTimestampExtractor}.
 * <p>
 * If a record has a negative (invalid) timestamp value, this extractor raises an exception.
 *
 * @see RobustConsumerRecordTimestampExtractor
 * @see InferringConsumerRecordTimestampExtractor
 * @see WallclockTimestampExtractor
 */
public class ConsumerRecordTimestampExtractor implements TimestampExtractor {

    /**
     * Extracts the embedded metadata timestamp from the given {@link ConsumerRecord}.
     *
     * @param record a data record
     * @param currentStreamsTime the current value of the internally tracked Streams time (could be -1 if unknown)
     * @return the embedded metadata timestamp of the given {@link ConsumerRecord}
     * @throws StreamsException if the embedded metadata timestamp is negative
     */
    @Override
    public long extract(final ConsumerRecord<Object, Object> record, final long currentStreamsTime) {
        final long timestamp = record.timestamp();

        if (timestamp < 0) {
            throw new StreamsException("Input record " + record + " has invalid (negative) timestamp. " +
                    "Possibly because a pre-0.10 producer client was used to write this record to Kafka without embedding a timestamp, " +
                    "or because the input topic was created before upgrading the Kafka cluster to 0.10+. " +
                    "Use a different TimestampExtractor to process this data.");
        }

        return timestamp;
    }

}

@@ -901,65 +901,8 @@
 ord;
-%0Aimport org.apache.kafka.streams.errors.StreamsException;
 %0A%0A/*
@@ -916,33 +916,24 @@
 trieves 
-embedded metadata
+built-in
  timesta
@@ -959,45 +959,10 @@
 ages
-.%0A * Embedded metadata timestamp was
  
+(
 intr
@@ -971,17 +971,16 @@
 uced in 
-%22
 KIP-32: 
@@ -1014,87 +1014,32 @@
 sage
-%22 for the new%0A * 0.10+ Kafka message format.%0A * %3Cp%3E%0A * Here, %22embedded metadata
+).%0A *%0A * Here, %22built-in
 %22 re
@@ -1163,25 +1163,17 @@
  message
- metadata
+s
  they se
@@ -1173,17 +1173,17 @@
 they sen
-d
+t
  to Kafk
@@ -1252,203 +1252,81 @@
 .%0A *
- %3Cp%3E
 %0A * If 
-the embedded metadata timestamp represents %3Ci%3ECreateTime%3C/i%3E (cf. Kafka broker setting%0A * %7B@code message.timestamp.type%7D and Kafka topic setting %7B@code log.message.timestamp.type%7D),%0A *
+%3Ci%3ECreateTime%3C/i%3E is used to define the built-in timestamps, using
  thi
@@ -1352,25 +1352,27 @@
 vely provide
-s
+%0A *
  %3Ci%3Eevent-ti
@@ -1388,19 +1388,16 @@
 mantics.
-%0A *
  If %3Ci%3EL
@@ -1425,63 +1425,30 @@
 sed 
-as broker/topic setting to define the embedded metadata
+to define the built-in
  tim
@@ -1459,17 +1459,17 @@
 mps,
-%0A *
  using
+%0A *
  thi
@@ -1532,28 +1532,24 @@
 emantics.%0A *
- %3Cp%3E
 %0A * If you n
@@ -1633,746 +1633,87 @@
 .%0A *
- %3Cp%3E%0A * If a record has a negative (invalid) timestamp value, this extractor raises an exception.%0A *%0A * @see RobustConsumerRecordTimestampExtractor%0A * @see InferringConsumerRecordTimestampExtractor%0A * @see WallclockTimestampExtractor%0A */%0Apublic class ConsumerRecordTimestampExtractor implements TimestampExtractor %7B%0A%0A    /**%0A     * Extracts the embedded metadata timestamp from the given %7B@link ConsumerRecord%7D.%0A     *%0A     * @param record a data record%0A     * @param currentStreamsTime the current value of the internally tracked Streams time (could be -1 if unknown)%0A     * @return the embedded metadata timestamp of the given %7B@link ConsumerRecord%7D%0A     * @throws StreamsException if the embedded metadata timestamp is negative%0A     */
+/%0Apublic class ConsumerRecordTimestampExtractor implements TimestampExtractor %7B
 %0A   
@@ -1747,22 +1747,16 @@
 extract(
-final 
 Consumer
@@ -1788,601 +1788,51 @@
 cord
-, final long currentStreamsTime) %7B%0A        final long timestamp = record.timestamp();%0A%0A        if (timestamp %3C 0) %7B%0A            throw new StreamsException(%22Input record %22 + record + %22 has invalid (negative) timestamp. %22 +%0A                    %22Possibly because a pre-0.10 producer client was used to write this record to Kafka without embedding a timestamp, %22 +%0A                    %22or because the input topic was created before upgrading the Kafka cluster to 0.10+. %22 +%0A                    %22Use a different TimestampExtractor to process this data.%22);%0A        %7D%0A%0A        return timestamp;%0A    %7D%0A
+) %7B%0A        return record.timestamp();%0A    %7D
 %0A%7D%0A

revert streams/src/main/java/org/apache/kafka/streams/processor/ConsumerRecordTimestampExtractor.java