Stop stream accepting messages when consuming in process

[irfanh94]

Summary by CodeRabbit

Release Notes

• Bug Fixes

  • Improved consumer stream management by implementing explicit pause and resume controls during batch processing for better resource efficiency.

• Tests

  • Added comprehensive test coverage for batch stream processing to validate pause/resume sequencing and proper control flow behavior.

_{✏️ Tip: You can customize this high-level summary in your review settings.}

[coderabbitai]

Walkthrough

The changes implement backpressure management for Kafka consumer batch processing by pausing the consumer stream before processing each batch and resuming it afterward. A unit test is added to verify the pause/resume behavior and correct invocation of the consume method.

Changes

Cohort / File(s)	Summary
Stream Backpressure Management packages/kafka/lib/AbstractKafkaConsumer.ts	Added explicit pause and resume calls around batch processing in the synchronous stream handling path to manage backpressure during batch consumption.
Unit Test packages/kafka/lib/AbstractKafkaConsumer.spec.ts	New test file verifying that handleSyncStreamBatch correctly pauses the consumer stream before consumption, resumes it after, and invokes consume exactly once.
Version Bump packages/kafka/package.json	Incremented package version from 0.8.1 to 0.8.2.

Estimated code review effort

🎯 2 (Simple) | ⏱️ ~10 minutes

• The test setup involves mocked dependencies and assertion patterns that should be reviewed for correctness against the actual stream pause/resume semantics.
• Verify that pause/resume calls are positioned correctly relative to batch consumption logic and don't introduce race conditions or deadlocks.

Possibly related PRs

• Waiting for a batch to be processed before accepting more messages #371 — Introduced handleSyncStreamBatch method for iterating and consuming batches; this PR adds pause/resume backpressure management around that batch handling.
• use for await instead of stream.on #360 — Modified synchronous stream handling in AbstractKafkaConsumer; this PR continues that pattern by adding controlled backpressure during batch consumption.

Suggested reviewers

• kibertoad
• kjamrog
• CarlosGamero

Poem

🐰 A stream pauses to catch its breath,
Before the batch flows through,
Then resumes its merry race—
Backpressure gentle and true! ✨

Pre-merge checks and finishing touches

✅ Passed checks (3 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately describes the main change: pausing the Kafka consumer stream during batch processing to prevent accepting messages while consuming is in progress.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

packages/kafka/lib/AbstractKafkaConsumer.ts (1)

227-238: Critical: Stream will remain paused if consume throws an error.

If consume throws an exception (line 232-235), the resume() call on line 236 will never execute, leaving the consumer stream permanently paused. This will halt all message processing until the application restarts.

Apply this diff to ensure the stream is always resumed:

  private async handleSyncStreamBatch(
    stream: KafkaMessageBatchStream<DeserializedMessage<SupportedMessageValues<TopicsConfig>>>,
  ): Promise<void> {
    for await (const messageBatch of stream) {
      this.consumerStream?.pause()
-     await this.consume(
-       messageBatch.topic,
-       messageBatch.messages as DeserializedMessage<SupportedMessageValues<TopicsConfig>>,
-     )
-     this.consumerStream?.resume()
+     try {
+       await this.consume(
+         messageBatch.topic,
+         messageBatch.messages as DeserializedMessage<SupportedMessageValues<TopicsConfig>>,
+       )
+     } finally {
+       this.consumerStream?.resume()
+     }
    }
  }

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between f2e4861 and 4b07a2a.

📒 Files selected for processing (3)

• packages/kafka/lib/AbstractKafkaConsumer.spec.ts (1 hunks)
• packages/kafka/lib/AbstractKafkaConsumer.ts (1 hunks)
• packages/kafka/package.json (1 hunks)

🧰 Additional context used

🧬 Code graph analysis (1)

packages/kafka/lib/AbstractKafkaConsumer.ts (1)

packages/kafka/lib/types.ts (2)

• DeserializedMessage (46-51)
• SupportedMessageValues (42-44)

⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (20)

• GitHub Check: general (22.x, @message-queue-toolkit/schemas) / build
• GitHub Check: general (20.x, @message-queue-toolkit/s3-payload-store) / build
• GitHub Check: general (22.x, @message-queue-toolkit/sqs) / build
• GitHub Check: general (20.x, @message-queue-toolkit/sqs) / build
• GitHub Check: general (24.x, @message-queue-toolkit/sns) / build
• GitHub Check: general (22.x, @message-queue-toolkit/outbox-core) / build
• GitHub Check: general (20.x, @message-queue-toolkit/outbox-core) / build
• GitHub Check: general (24.x, @message-queue-toolkit/amqp) / build
• GitHub Check: general (20.x, @message-queue-toolkit/metrics) / build
• GitHub Check: general (20.x, @message-queue-toolkit/gcs-payload-store) / build
• GitHub Check: general (22.x, @message-queue-toolkit/amqp) / build
• GitHub Check: general (22.x, @message-queue-toolkit/gcp-pubsub) / build
• GitHub Check: general (24.x, @message-queue-toolkit/sqs) / build
• GitHub Check: general (22.x, @message-queue-toolkit/sns) / build
• GitHub Check: general (24.x, @message-queue-toolkit/gcp-pubsub) / build
• GitHub Check: general (20.x, @message-queue-toolkit/gcp-pubsub) / build
• GitHub Check: general (20.x, @message-queue-toolkit/sns) / build
• GitHub Check: general (20.x, @message-queue-toolkit/redis-message-deduplication-store) / build
• GitHub Check: kafka (22.x) / build
• GitHub Check: kafka (24.x) / build

🔇 Additional comments (1)

packages/kafka/package.json (1)

3-3: LGTM! Appropriate version bump.

The patch version increment is suitable for the backpressure management feature being introduced.