fix: lz4 compression (#122)

* fix: lz4 compression

* fix: filter test node v20 zstd

Author: simone-sanfratello

docs/producer.md

@@ -20,13 +20,15 @@ Options:
 | `producerEpoch`         | `number`                                                           | Producer epoch.                                                                                                                                                      |
 | `idempotent`            | `boolean`                                                          | Idempotency of the producer.                                                                                                                                         |
 | `acks`                  | `number`                                                           | Acknowledgement to wait before returning.<br/><br/>Valid values are defined in the `ProduceAcks` enumeration.                                                        |
-| `compression`           | `string`                                                           | Compression algorithm to use before sending messages to the broker.<br/><br/>Valid values are exported in the `CompressionAlgorithms` enumeration.                   |
+| `compression`           | `string`                                                           | Compression algorithm to use before sending messages to the broker.<br/><br/>Valid values are: `snappy`, `lz4`, `gzip`, `zstd` |
 | `partitioner`           | `(message: MessageToProduce<Key, Value, HeaderKey, HeaderValue>) => number` | Partitioner to use to assign a partition to messages that lack it.<br/><br/>It is a function that receives a message and should return the partition number.         |
 | `repeatOnStaleMetadata` | `boolean`                                                          | Whether to retry a produce operation when the system detects outdated topic or broker information.<br/><br/>Default is `true`.                                       |
 | `serializers`           | `Serializers<Key, Value, HeaderKey, HeaderValue>`                  | Object that specifies which serialisers to use.<br/><br/>The object should only contain one or more of the `key`, `value`, `headerKey` and `headerValue` properties. |
 
 It also supports all the constructor options of `Base`.
 
+Notes: `zstd` is not available in node `v20`
+
 ## Basic Methods
 
 ### `send<Key, Value, HeaderKey, HeaderValue>(options[, callback])`

src/protocol/compression.ts

@@ -38,8 +38,8 @@ function ensureBuffer (data: Buffer | DynamicBuffer): Buffer {
 
 let snappyCompressSync: CompressionOperation | undefined
 let snappyDecompressSync: CompressionOperation | undefined
-let lz4CompressSync: CompressionOperation | undefined
-let lz4DecompressSync: CompressionOperation | undefined
+let lz4CompressFrameSync: CompressionOperation | undefined
+let lz4DecompressFrameSync: CompressionOperation | undefined
 
 function loadSnappy () {
   try {
@@ -57,8 +57,8 @@ function loadSnappy () {
 function loadLZ4 () {
   try {
     const lz4 = require('lz4-napi')
-    lz4CompressSync = lz4.compressSync
-    lz4DecompressSync = lz4.uncompressSync
+    lz4CompressFrameSync = lz4.compressFrameSync
+    lz4DecompressFrameSync = lz4.decompressFrameSync
     /* c8 ignore next 5 - In tests lz4-napi is always available */
   } catch (e) {
     throw new UnsupportedCompressionError(
@@ -112,19 +112,19 @@ export const compressionsAlgorithms = {
   lz4: {
     compressSync (data: Buffer | DynamicBuffer): Buffer {
       /* c8 ignore next 4 - In tests lz4-napi is always available */
-      if (!lz4CompressSync) {
+      if (!lz4CompressFrameSync) {
         loadLZ4()
       }
 
-      return lz4CompressSync!(ensureBuffer(data))
+      return lz4CompressFrameSync!(ensureBuffer(data))
     },
     decompressSync (data: Buffer | DynamicBuffer): Buffer {
       /* c8 ignore next 4 - In tests lz4-napi is always available */
-      if (!lz4DecompressSync) {
+      if (!lz4DecompressFrameSync) {
         loadLZ4()
       }
 
-      return lz4DecompressSync!(ensureBuffer(data))
+      return lz4DecompressFrameSync!(ensureBuffer(data))
     },
     bitmask: 3,
     available: true

test/clients/consumer/consumer.test.ts

@@ -2,6 +2,7 @@ import { deepStrictEqual, ok, strictEqual } from 'node:assert'
 import { randomUUID } from 'node:crypto'
 import { once } from 'node:events'
 import { test, type TestContext } from 'node:test'
+import zlib from 'node:zlib'
 import * as Prometheus from 'prom-client'
 import { type FetchResponse } from '../../../src/apis/consumer/fetch-v17.ts'
 import { kConnections, kFetchConnections, kOptions } from '../../../src/clients/base/base.ts'
@@ -1261,38 +1262,41 @@ test('fetch should retrieve messages from multiple batches', async t => {
   }
 })
 
-test.only('fetch should retrieve messages from multiple batches (compressed)', async t => {
-  const topic = await createTopic(t, true)
-  const producer = await createProducer(t)
+for (const compression of Object.values(CompressionAlgorithms)) {
+  test(`fetch should retrieve messages from multiple batches (compressed ${compression})`,
+    { skip: compression === 'snappy' || (compression === 'zstd' && !('zstdCompressSync' in zlib)) }, async t => {
+      const topic = await createTopic(t, true)
+      const producer = await createProducer(t)
 
-  const msg: MessageToProduce = { key: Buffer.from('test'), value: Buffer.from('test'), topic }
-  await producer.send({ acks: ProduceAcks.NO_RESPONSE, compression: CompressionAlgorithms.GZIP, messages: [msg] })
-  await producer.send({ acks: ProduceAcks.NO_RESPONSE, compression: CompressionAlgorithms.GZIP, messages: [msg] })
-  await producer.send({ acks: ProduceAcks.NO_RESPONSE, compression: CompressionAlgorithms.GZIP, messages: [msg] })
+      const msg: MessageToProduce = { key: Buffer.from('test'), value: Buffer.from('test'), topic }
+      await producer.send({ acks: ProduceAcks.NO_RESPONSE, compression, messages: [msg] })
+      await producer.send({ acks: ProduceAcks.NO_RESPONSE, compression, messages: [msg] })
+      await producer.send({ acks: ProduceAcks.NO_RESPONSE, compression, messages: [msg] })
 
-  const consumer = createConsumer(t, {})
+      const consumer = createConsumer(t, {})
 
-  const stream = await consumer.consume({
-    autocommit: true,
-    topics: [topic],
-    mode: MessagesStreamModes.EARLIEST,
-    fallbackMode: MessagesStreamFallbackModes.EARLIEST,
-    minBytes: 1024 * 1024,
-    maxBytes: 1024 * 1024,
-    maxWaitTime: 100
-  })
+      const stream = await consumer.consume({
+        autocommit: true,
+        topics: [topic],
+        mode: MessagesStreamModes.EARLIEST,
+        fallbackMode: MessagesStreamFallbackModes.EARLIEST,
+        minBytes: 1024 * 1024,
+        maxBytes: 1024 * 1024,
+        maxWaitTime: 100
+      })
 
-  let i = 0
-  for await (const message of stream) {
-    strictEqual(message.topic, topic)
-    strictEqual(message.key.toString(), 'test')
-    strictEqual(message.value.toString(), 'test')
+      let i = 0
+      for await (const message of stream) {
+        strictEqual(message.topic, topic)
+        strictEqual(message.key.toString(), 'test')
+        strictEqual(message.value.toString(), 'test')
 
-    if (++i === 3) {
-      break
-    }
-  }
-})
+        if (++i === 3) {
+          break
+        }
+      }
+    })
+}
 
 test('commit should commit offsets to Kafka and support diagnostic channels', async t => {
   const consumer = createConsumer(t)