fix: Avoid out of bounds error.

.gitignore

@@ -3,4 +3,5 @@ external/
 node_modules
 coverage
 .eslintcache
-.env
+.env
+.vscode

src/apis/consumer/fetch-v15.ts

@@ -1,4 +1,4 @@
-import { ResponseError } from '../../errors.ts'
+import { type GenericError, OutOfBoundsError, ResponseError } from '../../errors.ts'
 import { Reader } from '../../protocol/reader.ts'
 import { readRecordsBatch, type RecordsBatch } from '../../protocol/records.ts'
 import { Writer } from '../../protocol/writer.ts'
@@ -174,19 +174,28 @@ export function parseResponse (
             preferredReadReplica: r.readInt32()
           }
 
-          let recordsSize = r.readUnsignedVarInt()
-
           if (partition.errorCode !== 0) {
             errors.push([`/responses/${i}/partitions/${j}`, partition.errorCode])
           }
 
-          if (recordsSize > 1) {
-            recordsSize--
+          // We need to reduce the size by one to follow the COMPACT_RECORDS specification.
+          const recordsSize = r.readUnsignedVarInt() - 1
 
+          if (recordsSize > 0) {
             const recordsBatchesReader = Reader.from(r.buffer.subarray(r.position, r.position + recordsSize))
             partition.records = []
             do {
-              partition.records.push(readRecordsBatch(recordsBatchesReader))
+              try {
+                partition.records.push(readRecordsBatch(recordsBatchesReader))
+              } catch (err) {
+                // Contrary to other places in the protocol, records batches CAN BE truncated due to maxBytes argument.
+                // In that case we just ignore the error.
+                if ((err as GenericError).code === OutOfBoundsError.code) {
+                  break
+                }
+                /* c8 ignore next 3 - Hard to test */
+                throw err
+              }
             } while (recordsBatchesReader.position < recordsSize)
 
             r.skip(recordsSize)

src/apis/consumer/fetch-v16.ts

@@ -1,4 +1,4 @@
-import { ResponseError } from '../../errors.ts'
+import { type GenericError, OutOfBoundsError, ResponseError } from '../../errors.ts'
 import { Reader } from '../../protocol/reader.ts'
 import { readRecordsBatch, type RecordsBatch } from '../../protocol/records.ts'
 import { Writer } from '../../protocol/writer.ts'
@@ -174,19 +174,28 @@ export function parseResponse (
             preferredReadReplica: r.readInt32()
           }
 
-          let recordsSize = r.readUnsignedVarInt()
-
           if (partition.errorCode !== 0) {
             errors.push([`/responses/${i}/partitions/${j}`, partition.errorCode])
           }
 
-          if (recordsSize > 1) {
-            recordsSize--
+          // We need to reduce the size by one to follow the COMPACT_RECORDS specification.
+          const recordsSize = r.readUnsignedVarInt() - 1
 
+          if (recordsSize > 0) {
             const recordsBatchesReader = Reader.from(r.buffer.subarray(r.position, r.position + recordsSize))
             partition.records = []
             do {
-              partition.records.push(readRecordsBatch(recordsBatchesReader))
+              try {
+                partition.records.push(readRecordsBatch(recordsBatchesReader))
+              } catch (err) {
+                // Contrary to other places in the protocol, records batches CAN BE truncated due to maxBytes argument.
+                // In that case we just ignore the error.
+                if ((err as GenericError).code === OutOfBoundsError.code) {
+                  break
+                }
+                /* c8 ignore next 3 - Hard to test */
+                throw err
+              }
             } while (recordsBatchesReader.position < recordsSize)
 
             r.skip(recordsSize)

src/apis/consumer/fetch-v17.ts

@@ -1,4 +1,4 @@
-import { ResponseError } from '../../errors.ts'
+import { type GenericError, OutOfBoundsError, ResponseError } from '../../errors.ts'
 import { Reader } from '../../protocol/reader.ts'
 import { readRecordsBatch, type RecordsBatch } from '../../protocol/records.ts'
 import { Writer } from '../../protocol/writer.ts'
@@ -178,14 +178,24 @@ export function parseResponse (
             errors.push([`/responses/${i}/partitions/${j}`, partition.errorCode])
           }
 
-          // We need to reduce the size by one to follow the COMPACT_RECORDS specification
+          // We need to reduce the size by one to follow the COMPACT_RECORDS specification.
           const recordsSize = r.readUnsignedVarInt() - 1
 
           if (recordsSize > 0) {
             const recordsBatchesReader = Reader.from(r.buffer.subarray(r.position, r.position + recordsSize))
             partition.records = []
             do {
-              partition.records.push(readRecordsBatch(recordsBatchesReader))
+              try {
+                partition.records.push(readRecordsBatch(recordsBatchesReader))
+              } catch (err) {
+                // Contrary to other places in the protocol, records batches CAN BE truncated due to maxBytes argument.
+                // In that case we just ignore the error.
+                if ((err as GenericError).code === OutOfBoundsError.code) {
+                  break
+                }
+                /* c8 ignore next 3 - Hard to test */
+                throw err
+              }
             } while (recordsBatchesReader.position < recordsSize)
 
             r.skip(recordsSize)

src/errors.ts

@@ -11,6 +11,7 @@ export const errorCodes = [
   'PLT_KFK_AUTHENTICATION',
   'PLT_KFK_MULTIPLE',
   'PLT_KFK_NETWORK',
+  'PLT_KFK_OUT_OF_BOUNDS',
   'PLT_KFK_PROTOCOL',
   'PLT_KFK_RESPONSE',
   'PLT_KFK_TIMEOUT',
@@ -163,6 +164,14 @@ export class ProtocolError extends GenericError {
   }
 }
 
+export class OutOfBoundsError extends GenericError {
+  static code: ErrorCode = 'PLT_KFK_OUT_OF_BOUNDS'
+
+  constructor (message: string, properties: ErrorProperties = {}) {
+    super(OutOfBoundsError.code, message, { isOut: true, ...properties })
+  }
+}
+
 export class ResponseError extends MultipleErrors {
   static code: ErrorCode = 'PLT_KFK_RESPONSE'
 

src/protocol/dynamic-buffer.ts

@@ -1,4 +1,4 @@
-import { UserError } from '../errors.ts'
+import { OutOfBoundsError } from '../errors.ts'
 import { EMPTY_BUFFER, INT16_SIZE, INT32_SIZE, INT64_SIZE, INT8_SIZE } from './definitions.ts'
 import {
   BITS_8PLUS_MASK,
@@ -91,7 +91,7 @@ export class DynamicBuffer {
     }
 
     if (start < 0 || start > this.length || end > this.length) {
-      throw new UserError('Out of bounds.')
+      throw new OutOfBoundsError('Out of bounds.')
     }
 
     if (this.buffers.length === 0) {
@@ -128,13 +128,13 @@ export class DynamicBuffer {
     }
 
     if (start < 0 || start > this.length || end > this.length) {
-      throw new UserError('Out of bounds.')
+      throw new OutOfBoundsError('Out of bounds.')
     }
 
     if (this.buffers.length === 0) {
       return EMPTY_BUFFER
     } else if (this.buffers.length === 1) {
-      return this.buffers[0].slice(start, end)
+      return this.buffers[0].subarray(start, end)
     }
 
     let position = 0
@@ -176,7 +176,7 @@ export class DynamicBuffer {
 
   consume (offset: number): this {
     if (offset < 0 || offset > this.length) {
-      throw new UserError('Out of bounds.')
+      throw new OutOfBoundsError('Out of bounds.')
     }
 
     if (offset === 0) {
@@ -210,7 +210,7 @@ export class DynamicBuffer {
 
   get (offset: number): number {
     if (offset < 0 || offset >= this.length) {
-      throw new UserError('Out of bounds.')
+      throw new OutOfBoundsError('Out of bounds.')
     }
 
     const [finalIndex, current] = this.#findInitialBuffer(offset)
@@ -219,7 +219,7 @@ export class DynamicBuffer {
 
   readUInt8 (offset: number = 0): number {
     if (offset < 0 || offset >= this.length) {
-      throw new UserError('Out of bounds.')
+      throw new OutOfBoundsError('Out of bounds.')
     }
 
     const [finalIndex, current] = this.#findInitialBuffer(offset)
@@ -265,7 +265,7 @@ export class DynamicBuffer {
     let read = 0
 
     if (offset < 0 || offset >= this.length) {
-      throw new UserError('Out of bounds.')
+      throw new OutOfBoundsError('Out of bounds.')
     }
 
     // Find the initial buffer
@@ -294,7 +294,7 @@ export class DynamicBuffer {
     let read = 0
 
     if (offset < 0 || offset >= this.length) {
-      throw new UserError('Out of bounds.')
+      throw new OutOfBoundsError('Out of bounds.')
     }
 
     // Find the initial buffer
@@ -318,7 +318,7 @@ export class DynamicBuffer {
 
   readInt8 (offset: number = 0): number {
     if (offset < 0 || offset >= this.length) {
-      throw new UserError('Out of bounds.')
+      throw new OutOfBoundsError('Out of bounds.')
     }
 
     const [finalIndex, current] = this.#findInitialBuffer(offset)
@@ -681,7 +681,7 @@ export class DynamicBuffer {
 
   #readMultiple (index: number, length: number) {
     if (index < 0 || index + length > this.length) {
-      throw new UserError('Out of bounds.')
+      throw new OutOfBoundsError('Out of bounds.')
     }
 
     let [startOffset, current] = this.#findInitialBuffer(index)

test/apis/consumer/fetch-v15.test.ts

@@ -825,3 +825,106 @@ test('parseResponse parses record data', () => {
   // Verify value is a Buffer
   ok(Buffer.isBuffer(record.value))
 })
+
+test('parseResponse handles truncated records', () => {
+  // Create a response with records data
+  // First create a record batch
+  const timestamp = BigInt(Date.now())
+  const recordsBatch = Writer.create()
+    // Record batch structure
+    .appendInt64(0n) // firstOffset
+    .appendInt32(60) // length - this would be dynamically computed in real usage
+    .appendInt32(0) // partitionLeaderEpoch
+    .appendInt8(2) // magic (record format version)
+    .appendUnsignedInt32(0) // crc - would be computed properly in real code
+    .appendInt16(0) // attributes
+    .appendInt32(0) // lastOffsetDelta
+    .appendInt64(timestamp) // firstTimestamp
+    .appendInt64(timestamp) // maxTimestamp
+    .appendInt64(-1n) // producerId - not specified
+    .appendInt16(0) // producerEpoch
+    .appendInt32(0) // firstSequence
+    .appendInt32(1) // number of records
+    // Single record
+    .appendVarInt(8) // length of the record
+    .appendInt8(0) // attributes
+    .appendVarInt64(0n) // timestampDelta
+    .appendVarInt(0) // offsetDelta
+    .appendVarIntBytes(null) // key
+    .appendVarIntBytes(Buffer.from('test-value')) // value
+    .appendVarIntArray([], () => {}) // No headers
+    // Truncated batch
+    .appendInt64(0n) // firstOffset
+    .appendInt32(60) // length
+
+  // Now create the full response
+  const writer = Writer.create()
+    .appendInt32(0) // throttleTimeMs
+    .appendInt16(0) // errorCode (success)
+    .appendInt32(123) // sessionId
+    // Responses array - using tagged fields format
+    .appendArray(
+      [
+        {
+          topicId: '12345678-1234-1234-1234-123456789abc',
+          partitions: [
+            {
+              partitionIndex: 0,
+              errorCode: 0,
+              highWatermark: 100n,
+              lastStableOffset: 100n,
+              logStartOffset: 0n,
+              abortedTransactions: [],
+              preferredReadReplica: -1,
+              recordsBatch
+            }
+          ]
+        }
+      ],
+      (w, topic) => {
+        w.appendUUID(topic.topicId)
+          // Partitions array
+          .appendArray(topic.partitions, (w, partition) => {
+            w.appendInt32(partition.partitionIndex)
+              .appendInt16(partition.errorCode)
+              .appendInt64(partition.highWatermark)
+              .appendInt64(partition.lastStableOffset)
+              .appendInt64(partition.logStartOffset)
+              // Aborted transactions array (empty)
+              .appendArray(partition.abortedTransactions, () => {})
+              .appendInt32(partition.preferredReadReplica)
+
+              // Add records batch
+              .appendUnsignedVarInt(partition.recordsBatch.length + 1)
+              .appendFrom(partition.recordsBatch)
+          })
+      }
+    )
+    .appendInt8(0) // Root tagged fields
+
+  const response = parseResponse(1, 1, 17, Reader.from(writer))
+
+  // Verify the records were parsed correctly
+  ok(response.responses[0].partitions[0].records, 'Records should be defined')
+
+  const batch = response.responses[0].partitions[0].records[0]!
+  const record = batch.records[0]
+
+  deepStrictEqual(
+    {
+      firstOffset: batch.firstOffset,
+      recordsLength: batch.records.length,
+      offsetDelta: record.offsetDelta,
+      valueString: record.value.toString()
+    },
+    {
+      firstOffset: 0n,
+      recordsLength: 1,
+      offsetDelta: 0,
+      valueString: 'test-value'
+    }
+  )
+
+  // Verify value is a Buffer
+  ok(Buffer.isBuffer(record.value))
+})