Add highwatermark offset to eachBatch callback

[milindl]

Fixes #282.

[Copilot]

Pull Request Overview

This PR adds support for exposing the high watermark offset and lag information in the eachBatch callback, including updating tests and documentation.

• Updates consumer batch payload computation to include highWatermark, offsetLag, and offsetLagLow.
• Enhances tests to verify the correct reporting of these values for multiple partitions.
• Updates examples and the CHANGELOG to reflect these changes.

Reviewed Changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

File	Description
test/promisified/consumer/consumeMessages.spec.js	Adds a new test case to validate highWatermark and lag values.
lib/kafkajs/_consumer.js	Updates batch payload construction with highWatermark and lag calculations.
examples/typescript/kafkajs.ts	Adds configuration for auto.offset.reset.
CHANGELOG.md	Documents new features added in this release.

[Copilot]

Pull Request Overview

Adds support for reporting the high watermark and lag metrics in each batch callback and updates the consumer implementation and tests accordingly.

• Fetches and exposes highWatermark, offsetLag(), and offsetLagLow() in eachBatch
• Updates the internal consumer to call getWatermarkOffsets and calculate lag
• Adds tests and example usage demonstrating the new metrics

Reviewed Changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

File	Description
test/promisified/consumer/consumeMessages.spec.js	New test case validating highWatermark and lag values
lib/kafkajs/_consumer.js	Batch payload now fetches watermark offsets and computes lag
examples/typescript/kafkajs.ts	Example consumer config updated with offset reset option
CHANGELOG.md	Notes the new highWatermark and lag enhancements (v1.4.0)

Comments suppressed due to low confidence (3)

test/promisified/consumer/consumeMessages.spec.js:920

• The test waits for new messages but doesn’t assert that batch.highWatermark or lag metrics are updated after the second send; consider adding explicit assertions for the updated watermark and lag.

await waitFor(() => (messagesConsumed === (partition0ProducedMessages + messages1.length + messages2.length)), () => { }, 100);

lib/kafkajs/_consumer.js:866

• getWatermarkOffsets is likely asynchronous; you need to await its result and mark #createBatchPayload as async to ensure correct watermark values.

const watermarkOffsets = this.#internalClient.getWatermarkOffsets(topic, partition);

examples/typescript/kafkajs.ts:47

• [nitpick] The option 'auto.offset.reset' may not be recognized at this level; document or move it into the correct consumer config namespace or use fromBeginning: true instead.

'auto.offset.reset': 'earliest',

[sonarqube-confluent]

Analysis Details

0 Issues

• 0 Bugs
• 0 Vulnerabilities
• 0 Code Smells

Coverage and Duplications

• 86.70% Coverage (70.90% Estimated after merge)
• No duplication information (2.00% Estimated after merge)

Project ID: confluent-kafka-javascript

View in SonarQube

[emasab]

LGTM! Thanks for the implementation