KAFKA-8615: Change to track partition time breaks TimestampExtractor

streams/examples/src/main/java/org/apache/kafka/streams/examples/pageview/JsonTimestampExtractor.java

@@ -27,7 +27,7 @@
 public class JsonTimestampExtractor implements TimestampExtractor {
 
     @Override
-    public long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {
+    public long extract(final ConsumerRecord<Object, Object> record, final long partitionTime) {
         if (record.value() instanceof PageViewTypedDemo.PageView) {
             return ((PageViewTypedDemo.PageView) record.value()).timestamp;
         }

streams/src/main/java/org/apache/kafka/streams/processor/ExtractRecordMetadataTimestamp.java

@@ -50,15 +50,15 @@ abstract class ExtractRecordMetadataTimestamp implements TimestampExtractor {
      * Extracts the embedded metadata timestamp from the given {@link ConsumerRecord}.
      *
      * @param record a data record
-     * @param previousTimestamp the latest extracted valid timestamp of the current record's partition˙ (could be -1 if unknown)
+     * @param partitionTime the latest extracted valid timestamp of the current record's partition˙ (could be -1 if unknown)
      * @return the embedded metadata timestamp of the given {@link ConsumerRecord}
      */
     @Override
-    public long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {
+    public long extract(final ConsumerRecord<Object, Object> record, final long partitionTime) {
         final long timestamp = record.timestamp();
 
         if (timestamp < 0) {
-            return onInvalidTimestamp(record, timestamp, previousTimestamp);
+            return onInvalidTimestamp(record, timestamp, partitionTime);
         }
 
         return timestamp;

streams/src/main/java/org/apache/kafka/streams/processor/TimestampExtractor.java

@@ -46,8 +46,8 @@ public interface TimestampExtractor {
      *
      *
      * @param record a data record
-     * @param previousTimestamp the latest extracted valid timestamp of the current record's partition˙ (could be -1 if unknown)
+     * @param partitionTime the latest extracted valid timestamp of the current record's partition˙ (could be -1 if unknown)
      * @return the timestamp of the record
      */
-    long extract(ConsumerRecord<Object, Object> record, long previousTimestamp);
+    long extract(ConsumerRecord<Object, Object> record, long partitionTime);
 }

streams/src/main/java/org/apache/kafka/streams/processor/WallclockTimestampExtractor.java

@@ -38,11 +38,11 @@ public class WallclockTimestampExtractor implements TimestampExtractor {
      * Return the current wall clock time as timestamp.
      *
      * @param record a data record
-     * @param previousTimestamp the latest extracted valid timestamp of the current record's partition˙ (could be -1 if unknown)
+     * @param partitionTime the latest extracted valid timestamp of the current record's partition˙ (could be -1 if unknown)
      * @return the current wall clock time, expressed in milliseconds since midnight, January 1, 1970 UTC
      */
     @Override
-    public long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {
+    public long extract(final ConsumerRecord<Object, Object> record, final long partitionTime) {
         return System.currentTimeMillis();
     }
 }

streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordQueue.java

@@ -47,6 +47,7 @@ public class RecordQueue {
     private final ArrayDeque<ConsumerRecord<byte[], byte[]>> fifoQueue;
 
     private StampedRecord headRecord = null;
+    private long partitionTime = RecordQueue.UNKNOWN;
 
     private Sensor skipRecordsSensor;
 
@@ -153,6 +154,7 @@ public long timestamp() {
     public void clear() {
         fifoQueue.clear();
         headRecord = null;
+        partitionTime = RecordQueue.UNKNOWN;
     }
 
     private void updateHead() {
@@ -167,7 +169,7 @@ private void updateHead() {
 
             final long timestamp;
             try {
-                timestamp = timestampExtractor.extract(deserialized, timestamp());
+                timestamp = timestampExtractor.extract(deserialized, partitionTime);
             } catch (final StreamsException internalFatalExtractorException) {
                 throw internalFatalExtractorException;
             } catch (final Exception fatalUserException) {
@@ -189,6 +191,11 @@ private void updateHead() {
             }
 
             headRecord = new StampedRecord(deserialized, timestamp);
+
+            // update the partition timestamp if the current head record's timestamp has exceed its value
+            if (timestamp > partitionTime) {
+                partitionTime = timestamp;
+            }
         }
     }
 }

streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java

@@ -662,7 +662,7 @@ public Deserializer deserializer() {
     public static class MockTimestampExtractor implements TimestampExtractor {
 
         @Override
-        public long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {
+        public long extract(final ConsumerRecord<Object, Object> record, final long partitionTime) {
             return 0;
         }
     }

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

@@ -731,7 +731,7 @@ public static class CustomTimestampExtractor implements TimestampExtractor {
         private static final long DEFAULT_TIMESTAMP = 1000L;
 
         @Override
-        public long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {
+        public long extract(final ConsumerRecord<Object, Object> record, final long partitionTime) {
             if (record.value().toString().matches(".*@[0-9]+")) {
                 return Long.parseLong(record.value().toString().split("@")[1]);
             }

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordQueueTest.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.processor.internals;
 
+import java.util.ArrayList;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.metrics.Metrics;
@@ -38,7 +39,6 @@
 import org.apache.kafka.streams.state.StateSerdes;
 import org.apache.kafka.test.InternalMockProcessorContext;
 import org.apache.kafka.test.MockSourceNode;
-import org.apache.kafka.test.MockTimestampExtractor;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -53,7 +53,7 @@
 public class RecordQueueTest {
     private final Serializer<Integer> intSerializer = new IntegerSerializer();
     private final Deserializer<Integer> intDeserializer = new IntegerDeserializer();
-    private final TimestampExtractor timestampExtractor = new MockTimestampExtractor();
+    private final TimestampExtractor timestampExtractor = new RecordQueueTestTimestampExtractor();
     private final String[] topics = {"topic"};
 
     private final Sensor skippedRecordsSensor = new Metrics().sensor("skipped-records");
@@ -182,6 +182,29 @@ public void testTimeTracking() {
         assertEquals(4L, queue.timestamp());
     }
 
+    @Test
+    public void testTimestampExtractorPartitionTime() {
+
+        final RecordQueueTestTimestampExtractor testTimestampExtractor = (RecordQueueTestTimestampExtractor) timestampExtractor;
+
+        assertTrue(queue.isEmpty());
+        assertEquals(0, queue.size());
+        assertEquals(RecordQueue.UNKNOWN, queue.timestamp());
+
+        // add three 3 out-of-order records with timestamp 2, 1, 3, 4
+        final List<ConsumerRecord<byte[], byte[]>> list1 = Arrays.asList(
+            new ConsumerRecord<>("topic", 1, 2, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
+            new ConsumerRecord<>("topic", 1, 1, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
+            new ConsumerRecord<>("topic", 1, 3, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
+            new ConsumerRecord<>("topic", 1, 4, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue));
+
+        queue.addRawRecords(list1);
+        while (queue.poll() != null) {
+        }
+
+        assertEquals(testTimestampExtractor.observedPartitionTimes, new ArrayList<>(Arrays.asList(RecordQueue.UNKNOWN, 2L, 2L, 3L)));
+    }
+
     @Test(expected = StreamsException.class)
     public void shouldThrowStreamsExceptionWhenKeyDeserializationFails() {
         final byte[] key = Serdes.Long().serializer().serialize("foo", 1L);
@@ -253,4 +276,17 @@ public void shouldDropOnNegativeTimestamp() {
 
         assertEquals(0, queue.size());
     }
+
+    static class RecordQueueTestTimestampExtractor implements TimestampExtractor {
+        private final List<Long> observedPartitionTimes = new ArrayList<>();
+
+        public long extract(final ConsumerRecord<Object, Object> record, final long partitionTime) {
+            observedPartitionTimes.add(partitionTime);
+            return record.offset();
+        }
+
+        public List<Long> observedPartitionTimes() {
+            return observedPartitionTimes;
+        }
+    }
 }

streams/src/test/java/org/apache/kafka/test/MockTimestampExtractor.java

@@ -23,7 +23,7 @@
 public class MockTimestampExtractor implements TimestampExtractor {
 
     @Override
-    public long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {
+    public long extract(final ConsumerRecord<Object, Object> record, final long partitionTime) {
         return record.offset();
     }
 }