---
title: "consumer"
date: 0001-01-01
summary: "consumer #  Description #  The consumer processor is used to consume messages recorded in the queue without processing them. Its purpose is to provide an entry point for data consumption pipeline, which will be further processed by subsequent processors.
Configuration Example #  Here is a simple configuration example:
pipeline: - name: consume_queue_messages auto_start: true keep_running: true retry_delay_in_ms: 5000 processor: - consumer: consumer: fetch_max_messages: 1 max_worker_size: 200 num_of_slices: 1 idle_timeout_in_seconds: 30 queue_selector: keys: - email_messages processor: - xxx1: - xxx2: In the above example, it subscribes to and consumes the email_messages queue."
---


# consumer

## Description

The consumer processor is used to consume messages recorded in the `queue` without processing them. Its purpose is to provide an entry point for data consumption pipeline, which will be further processed by subsequent processors.

## Configuration Example

Here is a simple configuration example:

```
pipeline:
  - name: consume_queue_messages
    auto_start: true
    keep_running: true
    retry_delay_in_ms: 5000
    processor:
      - consumer:
          consumer:
            fetch_max_messages: 1
          max_worker_size: 200
          num_of_slices: 1
          idle_timeout_in_seconds: 30
          queue_selector:
            keys:
              - email_messages
          processor:
            - xxx1:
            - xxx2:
```

In the above example, it subscribes to and consumes the `email_messages` queue. The queue messages are stored in the context of the current pipeline. The consumer provides a `processor` parameter, which contains a series of processors that will be executed sequentially. If any processor encounters an error during execution, the `consumer` will exit without committing the batch of data.

## Parameter Description

| Name                  | Type   | Description                                                         |
| --------------------- | ------ | ------------------------------------------------------------------- |
| message_field         | string | The field name in the context where messages from the queue are stored. Default is `messages`. |
| max_worker_size       | int    | The maximum number of workers allowed to run simultaneously. Default is `10`. |
| num_of_slices         | int    | The number of parallel threads for consuming a single queue. Maximum slice size at runtime. |
| slices                | array  | Allowed slice numbers as an integer array.                         |
| queue_selector.labels | map    | Filter a group of queues to be consumed based on labels, similar to `queues` configuration. |
| queue_selector.ids    | array  | Specifies the UUIDs of the queues to be consumed, as a string array. |
| queue_selector.keys   | array  | Specifies the unique key paths of the queues to be consumed, as a string array. |
| queues                | map    | Filter a group of queues to be consumed based on labels, similar to `queue_selector.labels` configuration. |
| waiting_after         | array  | Whether to wait for specified queues to finish consumption before starting. UUIDs of the queues, as a string array. |
| idle_timeout_in_seconds | int  | Timeout duration for consuming queues. Default is `5` seconds. |
| detect_active_queue   | bool   | Whether to automatically detect new queues that meet the conditions. Default is `true`. |
| detect_interval       | int    | Time interval in milliseconds for automatically detecting new queues that meet the conditions. Default is `5000`. |
| quiet_detect_after_idle_in_ms | bool | Idle interval in milliseconds to exit automatic detection. Default is `30000`. |
| skip_empty_queue      | bool   | Whether to skip consuming empty queues. Default is `true`. |
| quit_on_eof_queue     | bool   | Automatically quit consuming when reaching the last message of a queue. Default is `true`. |
| consumer.source       | string | Consumer source.                                                   |
| consumer.id           | string | Unique identifier for the consumer.                                |
| consumer.name         | string | Consumer name.                                                     |
| consumer.group        | string | Consumer group name.                                               |
| consumer.fetch_min_bytes | int  | Minimum size in bytes for fetching messages. Default is `1`.       |
| consumer.fetch_max_bytes | int  | Maximum size in bytes for fetching messages. Default is `10485760`, which is 10MB. |
| consumer.fetch_max_messages | int | Maximum number of messages to fetch. Default is `1`.               |
| consumer.fetch_max_wait_ms | int | Maximum wait time in milliseconds for fetching messages. Default is `10000`. |
| consumer.eof_retry_delay_in_ms | int | Waiting time in milliseconds for retrying when reaching the end of a file. Default is `500`. |