Pull-based Ingestion API

Introduced 3.0

Pull-based ingestion enables OpenSearch to ingest data from streaming sources such as Apache Kafka or Amazon Kinesis. Unlike traditional ingestion methods where clients actively push data to OpenSearch through REST APIs, pull-based ingestion allows OpenSearch to control the data flow by retrieving data directly from streaming sources. This approach provides native backpressure handling, helping prevent server overload during traffic spikes. Pull-based ingestion guarantees at-least-once ingestion semantics and uses external versioning to ensure data consistency.

Prerequisites

Before using pull-based ingestion, ensure that the following prerequisites are met:

• Install an ingestion plugin for your streaming source using the command bin/opensearch-plugin install <plugin-name>. For more information, see Additional plugins. OpenSearch supports the following ingestion plugins:
  • ingestion-kafka
  • ingestion-kinesis (experimental)
• Configure pull-based ingestion during index creation. You cannot convert an existing push-based index to a pull-based one.

Creating an index for pull-based ingestion

To ingest data from a streaming source, first create an index with pull-based ingestion settings. The following request creates an index that pulls data from a Kafka topic in the segment replication mode. For other available modes, see Ingestion modes:

PUT /my-index
{
  "settings": {
    "ingestion_source": {
      "type": "kafka",
      "pointer.init.reset": "earliest",
      "param": {
        "topic": "test",
        "bootstrap_servers": "localhost:49353"
      }
    },
    "index.number_of_shards": 1,
    "index.number_of_replicas": 1,
    "index": {
      "replication.type": "SEGMENT"
    }
  },
  "mappings": {
    "properties": {
      "name": {
        "type": "text"
      },
      "age": {
        "type": "integer"
      }
    }
  }
}

Ingestion source settings

The ingestion_source settings control how OpenSearch pulls data from the streaming source. A poll is an operation in which OpenSearch actively requests a batch of data from the streaming source. The following table lists all settings that ingestion_source supports.

Dynamic settings can be updated using the Update Settings API without restarting the ingestion process. Static settings cannot be changed after index creation. For more information about static and dynamic settings, see Configuring OpenSearch.

Other settings

Pull-based ingestion supports the following OpenSearch settings.

Ingestion modes

Pull-based ingestion supports the following modes.

Segment replication mode

In segment replication mode, the primary shards ingest events from a streaming source and index the documents. The pull-based index is configured to use segment replication to copy over the segment files from primary to replica shards, as shown in the following image.

We recommend using this mode with remote-backed storage.

All-active mode

Enabling all-active mode allows both primary and replica shards to independently ingest and index events from the streaming source, as shown in the following image.

There is no replication or coordination between the shards, although replica shards may fetch segment files from the primary shard during bootstrapping if a local copy is unavailable. This mode is currently not supported with segment replication.

Stream position

When creating an index, you can specify where OpenSearch should start reading from the stream by configuring the pointer.init.reset and pointer.init.reset.value settings in the ingestion_source parameter. OpenSearch will resume reading from the last committed position for existing indexes.

The following table provides the valid pointer.init.reset values and their corresponding pointer.init.reset.value values.

Stream partitioning

When using partitioned streams (such as Kafka topics or Kinesis shards), note the following relationships between stream partitions and OpenSearch shards:

• OpenSearch shards map one-to-one to stream partitions.
• The number of index shards must be greater than or equal to the number of stream partitions.
• Extra shards beyond the number of partitions remain empty.
• Documents must be sent to the same partition for successful updates.

When using pull-based ingestion, traditional REST API–based ingestion is disabled for the index.

Updating the error policy

You can use the Update Settings API to dynamically update the error policy by setting index.ingestion_source.error_strategy to either DROP or BLOCK.

The following example demonstrates how to update the error policy:

PUT /my-index/_settings
{
  "index.ingestion_source.error_strategy": "DROP"
}

response = client.indices.put_settings(
  index = "my-index",
  body =   {
    "index.ingestion_source.error_strategy": "DROP"
  }
)

Message format

To be correctly processed by OpenSearch, messages in the streaming source must have the following format:

{"_id":"1", "_version":"1", "_source":{"name": "alice", "age": 30}, "_op_type": "index"}
{"_id":"2", "_version":"2", "_source":{"name": "alice", "age": 30}, "_op_type": "delete"}

Each data unit in the streaming source (Kafka message or Kinesis record) must include the following fields that specify how to create or modify an OpenSearch document. This is the default format supported by pull-based ingestion.

We recommend using the document _id field to prevent duplicates, since pull-based ingestion provides at-least-once ingestion semantics. If your producer cannot guarantee event ordering, also set _version field to ensure data consistency.

Alternatively, pull-based ingestion supports indexing raw payloads in append-only mode without transformations. To enable this behavior, set index.ingestion_source.mapper_type to raw_payload. Note that in this mode, the index mappings must conform to the message structure because dynamic mapping is not supported. When using raw_payload, you must provide raw JSON objects exactly as they appear in the incoming data stream, as shown in the following example:

{"name": "alice", "age": 30}
{"name": "bob", "age": 30}

Pull-based ingestion metrics

Pull-based ingestion provides metrics that can be used to monitor the ingestion process. The polling_ingest_stats metric is currently supported and is available at the shard level.

The following table lists the available polling_ingest_stats metrics.

To retrieve shard-level pull-based ingestion metrics, use the Nodes Stats API:

GET /_nodes/stats/indices?level=shards&pretty

response = client.nodes.info(
  metric = "indices",
  node_id = "stats",
  params = { "level": "shards", "pretty": "true" }
)

Limitations

The following limitations apply when using pull-based ingestion:

• Ingest pipelines are not compatible with pull-based ingestion.
• Dynamic mapping is not supported.
• Index rollover is not supported.
• Operation listeners are not supported.

Related documentation

• Pull-based ingestion management