fix: Do not try to clone functions. (#117)

* fix: Do not try to clone functions.

Signed-off-by: Paolo Insogna <[email protected]>

Author: ShogunPanda

.github/workflows/ci.yml

@@ -22,7 +22,7 @@ jobs:
             test-command: ci
     name: 'Node.js ${{ matrix.node-version }} / Confluent Kafka ${{ matrix.confluent-kafka-version }}'
     runs-on: ubuntu-latest
-    timeout-minutes: 10
+    timeout-minutes: 20
     steps:
       - name: Checkout
         uses: actions/checkout@v4

package.json

@@ -30,12 +30,12 @@
     "lint": "eslint --cache",
     "typecheck": "tsc -p . --noEmit",
     "format": "prettier -w benchmarks playground src test",
-    "test": "c8 -c test/config/c8-local.json node --env-file=test/config/env --no-warnings --test --test-timeout=300000 'test/**/*.test.ts'",
-    "test:ci": "c8 -c test/config/c8-ci.json node --env-file=test/config/env --no-warnings --test --test-timeout=300000 'test/**/*.test.ts'",
+    "test": "c8 -c test/config/c8-local.json node --env-file=test/config/env --no-warnings --test --test-timeout=600000 'test/**/*.test.ts'",
+    "test:ci": "c8 -c test/config/c8-ci.json node --env-file=test/config/env --no-warnings --test --test-timeout=600000 'test/**/*.test.ts'",
     "test:docker:up": "node scripts/docker.ts up -d --wait",
     "test:docker:down": "node scripts/docker.ts down",
     "ci": "npm run build && npm run lint && npm run test:ci",
-    "ci:v20": "npm run lint && rm -rf dist && tsc -p tsconfig.json && node dist/scripts/postbuild.js && cd dist && c8 -c ../test/config/c8-ci.json node --env-file=../test/config/env --no-warnings --test --test-timeout=300000",
+    "ci:v20": "npm run lint && rm -rf dist && tsc -p tsconfig.json && node dist/scripts/postbuild.js && cd dist && c8 -c ../test/config/c8-ci.json node --env-file=../test/config/env --no-warnings --test --test-timeout=600000",
     "prepublishOnly": "npm run build && npm run lint",
     "postpublish": "git push origin && git push origin -f --tags",
     "generate:apis": "node scripts/generate-apis.ts",
@@ -81,4 +81,4 @@
   "engines": {
     "node": ">= 20.19.4 || >= 22.18.0 || >= 24.6.0"
   }
-}
+}

src/clients/consumer/messages-stream.ts

@@ -70,8 +70,18 @@ export class MessagesStream<Key, Value, HeaderKey, HeaderValue> extends Readable
     consumer: Consumer<Key, Value, HeaderKey, HeaderValue>,
     options: ConsumeOptions<Key, Value, HeaderKey, HeaderValue>
   ) {
-    const { autocommit, mode, fallbackMode, maxFetches, offsets, deserializers, onCorruptedMessage, ...otherOptions } =
-      options
+    const {
+      autocommit,
+      mode,
+      fallbackMode,
+      maxFetches,
+      offsets,
+      deserializers,
+      onCorruptedMessage,
+      // The options below are only destructured to avoid being part of structuredClone below
+      partitionAssigner: _partitionAssigner,
+      ...otherOptions
+    } = options
 
     if (offsets && mode !== MessagesStreamModes.MANUAL) {
       throw new UserError('Cannot specify offsets when the stream mode is not MANUAL.')

test/clients/consumer/consumer.test.ts

@@ -1261,7 +1261,7 @@ test('fetch should retrieve messages from multiple batches', async t => {
   }
 })
 
-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)
 
@@ -1276,7 +1276,10 @@ test('fetch should retrieve messages from multiple batches', async t => {
     autocommit: true,
     topics: [topic],
     mode: MessagesStreamModes.EARLIEST,
-    fallbackMode: MessagesStreamFallbackModes.EARLIEST
+    fallbackMode: MessagesStreamFallbackModes.EARLIEST,
+    minBytes: 1024 * 1024,
+    maxBytes: 1024 * 1024,
+    maxWaitTime: 100
   })
 
   let i = 0