Support LOG_APPEND_TIME record timestamps

src/broker/__tests__/fetch.spec.js

@@ -74,6 +74,7 @@ describe('Broker > Fetch', () => {
     partitionLeaderEpoch: expect.any(Number),
     producerEpoch: 0,
     producerId: '-1',
+    timestampType: 0,
     ...options,
   })
 

src/protocol/recordBatch/fixtures/v0_recordbatch.json

@@ -0,0 +1 @@
+{"type":"Buffer","data":[0,0,0,0,0,0,0,0,0,0,0,113,0,0,0,0,2,228,195,36,165,0,9,0,0,0,0,0,0,1,115,237,245,255,167,0,0,1,115,237,245,255,201,255,255,255,255,255,255,255,255,0,0,0,0,0,0,0,0,0,1,31,139,8,0,0,0,0,0,0,19,11,99,96,96,224,202,78,173,212,53,112,40,75,204,41,77,213,53,208,53,50,48,50,208,53,176,208,53,52,9,49,52,183,50,52,182,50,176,208,51,179,48,138,98,0,0,181,227,241,167,44,0,0,0]}

src/protocol/recordBatch/record/v0/decoder.js

@@ -1,5 +1,6 @@
 const Long = require('long')
 const HeaderDecoder = require('../../header/v0/decoder')
+const TimestampTypes = require('../../../timestampTypes')
 
 /**
  * v0
@@ -16,13 +17,23 @@ const HeaderDecoder = require('../../header/v0/decoder')
  */
 
 module.exports = (decoder, batchContext = {}) => {
-  const { firstOffset, firstTimestamp, magicByte, isControlBatch = false } = batchContext
+  const {
+    firstOffset,
+    firstTimestamp,
+    magicByte,
+    isControlBatch = false,
+    timestampType,
+    maxTimestamp,
+  } = batchContext
   const attributes = decoder.readInt8()
 
   const timestampDelta = decoder.readVarLong()
-  const timestamp = Long.fromValue(firstTimestamp)
-    .add(timestampDelta)
-    .toString()
+  const timestamp =
+    timestampType === TimestampTypes.LOG_APPEND_TIME && maxTimestamp
+      ? maxTimestamp
+      : Long.fromValue(firstTimestamp)
+          .add(timestampDelta)
+          .toString()
 
   const offsetDelta = decoder.readVarInt()
   const offset = Long.fromValue(firstOffset)

src/protocol/recordBatch/record/v0/decoder.spec.js

@@ -1,5 +1,6 @@
 const Decoder = require('../../../decoder')
 const recordDecoder = require('./decoder')
+const TimestampTypes = require('../../../timestampTypes')
 
 describe('Protocol > RecordBatch > Record > v0', () => {
   test('decodes', async () => {
@@ -28,6 +29,22 @@ describe('Protocol > RecordBatch > Record > v0', () => {
     })
   })
 
+  test('uses record batch maxTimestamp when topic is configured with timestamp type LOG_APPEND_TIME', async () => {
+    const decoder = new Decoder(Buffer.from(require('../../fixtures/v0_record.json')))
+    const batchContext = {
+      firstOffset: '0',
+      firstTimestamp: '1509827900073',
+      magicByte: 2,
+      maxTimestamp: '1597425188809',
+      timestampType: TimestampTypes.LOG_APPEND_TIME,
+    }
+
+    const decoded = recordDecoder(decoder, batchContext)
+
+    expect(decoded.batchContext).toEqual(batchContext)
+    expect(decoded.timestamp).toEqual(batchContext.maxTimestamp)
+  })
+
   test('decodes control record', async () => {
     const decoder = new Decoder(Buffer.from(require('../../fixtures/v0_record.json')))
 

src/protocol/recordBatch/v0/decoder.js

@@ -2,7 +2,9 @@ const Decoder = require('../../decoder')
 const { KafkaJSPartialMessageError } = require('../../../errors')
 const { lookupCodecByRecordBatchAttributes } = require('../../message/compression')
 const RecordDecoder = require('../record/v0/decoder')
+const TimestampTypes = require('../../timestampTypes')
 
+const TIMESTAMP_TYPE_FLAG_MASK = 0x8
 const TRANSACTIONAL_FLAG_MASK = 0x10
 const CONTROL_FLAG_MASK = 0x20
 
@@ -58,6 +60,11 @@ module.exports = async fetchDecoder => {
 
   const inTransaction = (attributes & TRANSACTIONAL_FLAG_MASK) > 0
   const isControlBatch = (attributes & CONTROL_FLAG_MASK) > 0
+  const timestampType =
+    (attributes & TIMESTAMP_TYPE_FLAG_MASK) > 0
+      ? TimestampTypes.LOG_APPEND_TIME
+      : TimestampTypes.CREATE_TIME
+
   const codec = lookupCodecByRecordBatchAttributes(attributes)
 
   const recordContext = {
@@ -71,6 +78,7 @@ module.exports = async fetchDecoder => {
     producerEpoch,
     firstSequence,
     maxTimestamp,
+    timestampType,
   }
 
   const records = await decodeRecords(codec, decoder, { ...recordContext, magicByte })

src/protocol/recordBatch/v0/decoder.spec.js

@@ -0,0 +1,24 @@
+const Decoder = require('../../decoder')
+const recordBatchDecoder = require('./decoder')
+
+describe('Protocol > RecordBatch > v0', () => {
+  test('decodes', async () => {
+    const decoder = new Decoder(Buffer.from(require('../fixtures/v0_recordbatch.json')))
+    const decoded = await recordBatchDecoder(decoder)
+
+    expect(decoded).toEqual({
+      firstOffset: '0',
+      firstTimestamp: '1597425188775',
+      partitionLeaderEpoch: 0,
+      inTransaction: false,
+      isControlBatch: false,
+      lastOffsetDelta: 0,
+      producerId: '-1',
+      producerEpoch: 0,
+      firstSequence: 0,
+      maxTimestamp: '1597425188809',
+      timestampType: 1,
+      records: expect.any(Object),
+    })
+  })
+})

src/protocol/requests/fetch/v10/response.spec.js

@@ -13,6 +13,7 @@ describe('Protocol > Requests > Fetch > v10', () => {
     partitionLeaderEpoch: expect.any(Number),
     producerEpoch: expect.any(Number),
     producerId: expect.any(String),
+    timestampType: expect.any(Number),
   }
 
   test('response', async () => {

src/protocol/requests/fetch/v11/response.spec.js

@@ -13,6 +13,7 @@ describe('Protocol > Requests > Fetch > v11', () => {
     partitionLeaderEpoch: expect.any(Number),
     producerEpoch: expect.any(Number),
     producerId: expect.any(String),
+    timestampType: expect.any(Number),
   }
 
   test('response', async () => {

src/protocol/requests/fetch/v4/response.spec.js

@@ -14,6 +14,7 @@ describe('Protocol > Requests > Fetch > v4', () => {
     partitionLeaderEpoch: expect.any(Number),
     producerEpoch: expect.any(Number),
     producerId: expect.any(String),
+    timestampType: expect.any(Number),
   }
 
   test('response', async () => {

src/protocol/requests/fetch/v5/response.spec.js

@@ -13,6 +13,7 @@ describe('Protocol > Requests > Fetch > v5', () => {
     partitionLeaderEpoch: expect.any(Number),
     producerEpoch: expect.any(Number),
     producerId: expect.any(String),
+    timestampType: expect.any(Number),
   }
 
   test('response', async () => {

src/protocol/requests/fetch/v6/response.spec.js

@@ -13,6 +13,7 @@ describe('Protocol > Requests > Fetch > v6', () => {
     partitionLeaderEpoch: expect.any(Number),
     producerEpoch: expect.any(Number),
     producerId: expect.any(String),
+    timestampType: expect.any(Number),
   }
 
   test('response', async () => {

src/protocol/requests/fetch/v7/response.spec.js

@@ -13,6 +13,7 @@ describe('Protocol > Requests > Fetch > v7', () => {
     partitionLeaderEpoch: expect.any(Number),
     producerEpoch: expect.any(Number),
     producerId: expect.any(String),
+    timestampType: expect.any(Number),
   }
 
   test('response', async () => {

src/protocol/requests/fetch/v8/response.spec.js

@@ -13,6 +13,7 @@ describe('Protocol > Requests > Fetch > v8', () => {
     partitionLeaderEpoch: expect.any(Number),
     producerEpoch: expect.any(Number),
     producerId: expect.any(String),
+    timestampType: expect.any(Number),
   }
 
   test('response', async () => {

src/protocol/requests/fetch/v9/response.spec.js

@@ -13,6 +13,7 @@ describe('Protocol > Requests > Fetch > v9', () => {
     partitionLeaderEpoch: expect.any(Number),
     producerEpoch: expect.any(Number),
     producerId: expect.any(String),
+    timestampType: expect.any(Number),
   }
 
   test('response', async () => {

src/protocol/timestampTypes.js

@@ -0,0 +1,15 @@
+/**
+ * Enum for timestamp types
+ * @readonly
+ * @enum {TimestampType}
+ */
+module.exports = {
+  // Timestamp type is unknown
+  NO_TIMESTAMP: -1,
+
+  // Timestamp relates to message creation time as set by a Kafka client
+  CREATE_TIME: 0,
+
+  // Timestamp relates to the time a message was appended to a Kafka log
+  LOG_APPEND_TIME: 1,
+}