Connect
Also in Cloud

memory

Stores consumed messages in memory and acknowledges them at the input level. During shutdown Redpanda Connect will make a best attempt at flushing all remaining messages before exiting cleanly.

This buffer is appropriate when consuming messages from inputs that do not gracefully handle back pressure and where delivery guarantees aren’t critical.

This buffer has a configurable limit, where consumption will be stopped with back pressure upstream if the total size of messages in the buffer reaches this amount. Since this calculation is only an estimate, and the real size of messages in RAM is always higher, it is recommended to set the limit significantly below the amount of RAM available.

Delivery guarantees

This buffer intentionally weakens the delivery guarantees of the pipeline and therefore should never be used in places where data loss is unacceptable.

Batching

It is possible to batch up messages sent from this buffer using a batch policy.

  • Common

  • Advanced

buffer:
  memory:
    limit: 524288000
    batch_policy:
      enabled: false
      count: 0
      byte_size: 0
      period: ""
      check: ""
buffer:
  memory:
    limit: 524288000
    batch_policy:
      enabled: false
      count: 0
      byte_size: 0
      period: ""
      check: ""
      processors: [] # No default (optional)

Fields

batch_policy

Optionally configure a policy to flush buffered messages in batches.

Type: object

batch_policy.byte_size

The maximum total size (in bytes) that a batch can reach before it is flushed. When the combined size of all messages in the batch reaches or exceeds this limit, the batch is immediately sent to the next stage (such as a processor or output).

Set to 0 to disable size-based batching. When disabled, messages are flushed based on other conditions (such as count or period).

Type: int

Default: 0

batch_policy.check

A Bloblang query that returns a boolean value indicating whether a message should end a batch.

Type: string

Default: ""

# Examples:
check: this.type == "end_of_transaction"

batch_policy.count

The number of messages at which the batch is flushed. Set to 0 to disable count-based batching.

Type: int

Default: 0

batch_policy.enabled

Whether to batch messages as they are flushed.

Type: bool

Default: false

batch_policy.period

The length of time after which an incomplete batch is flushed regardless of its size. This field accepts Go duration format strings such as 100ms, 1s, or 5s. Supported time units are ns, us, ms, s, m, and h.

Type: string

Default: ""

# Examples:
period: 1s

# ---

period: 1m

# ---

period: 500ms

batch_policy.processors[]

A list of processors to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. All resulting messages are flushed as a single batch, so splitting the batch into smaller batches with these processors has no effect.

Type: array<processor>

# Examples:
processors:
  - archive:
      format: concatenate


# ---

processors:
  - archive:
      format: lines


# ---

processors:
  - archive:
      format: json_array

limit

The maximum buffer size (in bytes) to allow before applying backpressure upstream.

Type: int

Default: 524288000