[graytaylor0]

Description

The Kafka buffer has kafka consumers that poll from kafka and write to an in memory bounded buffer. Before this change, the buffer and batch size (the amount of records that are pulled from this buffer by the processor workers at one time) were not configurable and set to 1,000,000 buffer size and 250,000 batch size.

This change make it so that these values can be configured based on the user's record size or compute that is being run on.

buffer:
  kafka:
    topics:
      - name: my-topic
        group_id: my-group
    bounded_buffer_size: 400000   # was hardcoded at 1,000,000
    bounded_batch_size: 100000    # was hardcoded at 250,000

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

Check List

• New functionality includes testing.
• New functionality has a documentation issue. Please link to it in this PR.
  • New functionality has javadoc added
• Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 25168e4)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Default values inconsistent with PR description The PR description states the new defaults should be 400,000 for bounded_buffer_size and 100,000 for bounded_batch_size (lowered from the hardcoded 1,000,000/250,000). However, DEFAULT_BOUNDED_BUFFER_SIZE is set to 1000000 and DEFAULT_BOUNDED_BATCH_SIZE is set to 250000, which keeps the previous values rather than lowering them as advertised in both the PR title and body. Either the defaults or the description should be corrected. static final int DEFAULT_BOUNDED_BUFFER_SIZE = 1000000; static final int DEFAULT_BOUNDED_BATCH_SIZE = 250000; Missing input validation boundedBufferSize and boundedBatchSize are user-configurable int fields with no validation (e.g., @Min/@Positive). A user supplying 0, a negative value, or a batch size larger than the buffer size will cause BlockingBuffer construction to fail or produce misleading runtime errors at pipeline startup. Consider adding Bean Validation annotations and a cross-field check that boundedBatchSize <= boundedBufferSize. @JsonProperty("bounded_buffer_size") private int boundedBufferSize = DEFAULT_BOUNDED_BUFFER_SIZE; @JsonProperty("bounded_batch_size") private int boundedBatchSize = DEFAULT_BOUNDED_BATCH_SIZE;

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 25168e4

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Validate configurable buffer and batch sizes The PR description mentions lowering default values, but the defaults here (1,000,000 and 250,000) are the same as the previously hardcoded values. Additionally, there is no validation that boundedBatchSize is ≤ boundedBufferSize or that both are positive — a misconfiguration (e.g., batch size larger than buffer capacity, or zero/negative) will cause BlockingBuffer construction or runtime failures. Consider adding @Min/@AssertTrue validation. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/buffer/KafkaBufferConfig.java [71-75] @JsonProperty("bounded_buffer_size") +@Min(1) private int boundedBufferSize = DEFAULT_BOUNDED_BUFFER_SIZE; @JsonProperty("bounded_batch_size") +@Min(1) private int boundedBatchSize = DEFAULT_BOUNDED_BATCH_SIZE; +@AssertTrue(message = "bounded_batch_size must be less than or equal to bounded_buffer_size") +boolean isBatchSizeValid() { + return boundedBatchSize <= boundedBufferSize; +} + Suggestion importance[1-10]: 6 __ Why: Adding validation for boundedBufferSize and boundedBatchSize is a reasonable defensive measure since misconfiguration could cause runtime failures. However, the impact is moderate as default values are provided and improper values would likely fail quickly.	Low

---

Previous suggestions

Suggestions up to commit b26e400

Category	Suggestion	Impact
General	Validate new configurable size properties The new configurable boundedBufferSize and boundedBatchSize fields have no validation. Invalid values (zero, negative, or boundedBatchSize > boundedBufferSize) will cause BlockingBuffer construction to fail at runtime or lead to unexpected behavior. Add @Positive (or @Min(1)) and an @AssertTrue cross-field validation to ensure batch size does not exceed buffer size. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/buffer/KafkaBufferConfig.java [71-75] @JsonProperty("bounded_buffer_size") +@Positive private int boundedBufferSize = DEFAULT_BOUNDED_BUFFER_SIZE; @JsonProperty("bounded_batch_size") +@Positive private int boundedBatchSize = DEFAULT_BOUNDED_BATCH_SIZE; +@AssertTrue(message = "bounded_batch_size must be less than or equal to bounded_buffer_size") +boolean isBatchSizeValid() { + return boundedBatchSize <= boundedBufferSize; +} + Suggestion importance[1-10]: 6 __ Why: Adding validation for boundedBufferSize and boundedBatchSize is a reasonable improvement to prevent misconfiguration, though it addresses a minor edge case rather than a critical issue.	Low

Suggestions up to commit 4e30a62

Category	Suggestion	Impact
General	Validate configurable buffer and batch sizes These user-facing configuration fields lack validation. A user could configure zero, negative, or inconsistent values (e.g., boundedBatchSize > boundedBufferSize), which would cause BlockingBuffer to throw at construction or behave incorrectly. Add bean-validation constraints such as @Positive and ideally a cross-field check to ensure boundedBatchSize <= boundedBufferSize. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/buffer/KafkaBufferConfig.java [71-75] @JsonProperty("bounded_buffer_size") +@Positive private int boundedBufferSize = DEFAULT_BOUNDED_BUFFER_SIZE; @JsonProperty("bounded_batch_size") +@Positive private int boundedBatchSize = DEFAULT_BOUNDED_BATCH_SIZE; Suggestion importance[1-10]: 6 __ Why: Adding @Positive validation on user-configurable boundedBufferSize and boundedBatchSize is a reasonable improvement to prevent invalid configurations causing runtime failures in BlockingBuffer. However, the impact is moderate as these are optional config fields with sensible defaults.	Low

[github-actions]

Persistent review updated to latest commit b26e400

[dlvenable]

We should keep the same defaults so that an in-place version update doesn't have a performance impact.

[github-actions]

Persistent review updated to latest commit 25168e4

[dlvenable]

Thanks @graytaylor0 !