FEAT: Add max_records to kafka task

[Divyanshu Tiwari (divyanshu-tiwari)]

Summary

• Adds a max_records field to the kafka read task that stops the reader after N records have been forwarded downstream.
• Independent from end_after (wall-clock) and retry_limit (idle-based); the three can be combined.
• In group consumer mode, offsets up to the last forwarded record are committed on shutdown via the deferred c.Close().
• README updated with the new field, an example, and a behavior note; new test pipeline test/pipelines/kafka_read_max_records.yaml added.

Test plan

• Run test/pipelines/kafka_read_max_records.yaml against a populated topic and verify exactly 10 records are emitted before the reader stops.
• Re-run the same pipeline in group-consumer mode and confirm offsets for the consumed records are committed (subsequent runs resume past them).
• Run against an empty/low-traffic topic and confirm the reader still exits cleanly via end_after / retry_limit when max_records cannot be reached.
• go build ./... is clean.

🤖 Generated with Claude Code

[Divyanshu Tiwari (divyanshu-tiwari)]

Copilot resolve the merge conflicts in this pull request

[Copilot]

Pull request overview

Adds a max_records read-mode limit to the Kafka pipeline task so the consumer can stop cleanly after forwarding a fixed number of messages, independent of existing end_after (wall-clock) and retry_limit (idle/error-based) stop conditions.

Changes:

• Introduces max_records configuration and enforces the stop condition in the Kafka consumer read loop.
• Updates Kafka task README with the new field, examples, and behavior notes.
• Adds a new test pipeline YAML demonstrating max_records + end_after.

Reviewed changes

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

File	Description
test/pipelines/kafka_read_max_records.yaml	Adds an example pipeline that stops after forwarding 10 Kafka records (with end_after safety net).
internal/pkg/pipeline/task/kafka/README.md	Documents max_records, including usage example and interaction with end_after/retry_limit.
internal/pkg/pipeline/task/kafka/kafka.go	Implements max_records counter and early exit after N forwarded records.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Yash Shrivastava (alephys26)]

This is problematic, the kafka consumer would fetch records in batch, but would not forward all of them downstream. This will lead to dropped records.
Add flush of all records here as well as in end_after part. So that we ensure at_least_once delivery.

[Yash Shrivastava (alephys26)]

Correction: This is valid for the kafka task. The ctx cancellation happens before a new read and the commit happens before the count increment. Both cases are valid.

The only pain point now is the failure of record processing in any downstream task. If it fails there we will drop the messages, since the commits are already there on kafka.

[Divyanshu Tiwari (divyanshu-tiwari)]

Correction: This is valid for the kafka task. The ctx cancellation happens before a new read and the commit happens before the count increment. Both cases are valid.

The only pain point now is the failure of record processing in any downstream task. If it fails there we will drop the messages, since the commits are already there on kafka.

It's a known issue, same as the SQS.