[RFC] Pull-Based Ingestion Support in the OpenSearch Sink

Motivation

OpenSearch now supports pull-based ingestion, where instead of receiving documents via the Bulk API, OpenSearch pulls documents from a streaming source such as Kafka. This eliminates the overhead of HTTP-based bulk requests and leverages OpenSearch's internal ingestion engine which writes directly to Lucene (bypassing the translog).

Data Prepper's opensearch sink currently uses push-based bulk ingestion exclusively.
Adding pull-based ingestion as a second mode allows users to choose the ingestion strategy that best fits their deployment. The two ingestion approaches will share much of the same concepts and code — index management, document ID resolution, action resolution, DLQ handling — but differ in how documents reach OpenSearch.

Using pull-based ingestion can offer a few advantages for the community:

• The streams used by pull-based ingestion can be replicated to support multiple regions.
• This can reduce the load on the OpenSearch data nodes.
• The version data from various pull-based sources (e.g. DynamoDB) can be embedded into the ingestion stream to handle conflicts.

Additionally, pull-based ingestion can be complicated since it requires innate knowledge of the OpenSearch cluster such as shard counts. Using Data Prepper can simplify the approach.

Configuration

opensearch:
  hosts: ["https://myopensearch.org:9200"]
  index: "my-index"
  document_id: "${/doc_id}"              # required in Phase 1, optional later
  document_version: "${/timestamp}"      # required in Phase 1, optional later
  action: "index"                        # optional, default "index"; only index, create, delete supported
  pull_indexing:
    engine:
      kafka:
        bootstrap_servers: ["kafka-1:9092", "kafka-2:9092"]
        topic:
          name: "my-index-ingestion"     # optional, defaults to index name
        authentication:                  # optional
          sasl:
            plaintext:
              username: "user"
              password: "pass"
        encryption:                      # optional
          type: "ssl"
        producer_properties:             # optional tuning
          compression_type: "zstd"
          batch_size: 65536
          linger_ms: 5
          buffer_memory: "50mb"

When pull_indexing is present, the sink will ingest using pull-based ingestion. When it is absent, behavior is unchanged (bulk ingestion).

The engine field uses Data Prepper's PluginModel pattern, making the streaming transport pluggable.
The kafka plugin is the first implementation, but the design supports future engines (e.g., Kinesis) without changing the core sink.

The configuration structure for the kafka plugin matches the existing Kafka sink plugin (kafka-plugins), so that we can reuse the same configuration classes. The authentication block supports the same options: sasl.plaintext, sasl.scram, sasl.oauth, sasl.aws_msk_iam. The topic block supports name, create_topic, replication_factor, and other topic-level settings. The producer_properties block supports all Kafka producer tuning options.

Constraints When Pull Indexing Is Enabled

• Static index required. The index must be a literal string, not a Data Prepper expression. Dynamic index support is deferred to a future phase.
• No routing. The routing and routing_field options are not supported. Partition assignment is based solely on document ID (see Partition Assignment). This could be changed later.
• Supported actions only. Only index, create, and delete are allowed. Pull-based ingestion does not support partial updates, so update and upsert actions are rejected at configuration validation time.
• Index must not already exist. During initialization, the sink creates the OpenSearch index with the ingestion_source settings required for pull-based ingestion. If the index already exists, initialization fails. This avoids the complexity of validating that an existing index has the correct ingestion source, replication type, and shard count. We can add support for existing indexes later.

Design

Component Overview

OpenSearchSink
├── IndexManager          (shared: index setup, template management)
├── EventActionResolver   (shared: action resolution)
├── Ingester (interface)
│   ├── BulkIngester      (existing: Bulk API path)
│   └── PullIngester      (new: streaming path)
│       └── PullEngine (interface)
│           └── KafkaPullEngine (in opensearch-pull-kafka project)

KafkaPullEngine (in opensearch-pull-kafka project)
└── PullEngine (implements)

Ingester Selection

OpenSearchSink selects the ingester based on configuration:

// In OpenSearchSink constructor
if (openSearchSinkConfig.getPullIndexing() != null) {
    PullEngine engine = pluginFactory.loadPlugin(
        PullEngine.class, openSearchSinkConfig.getPullIndexing().getEngine());
    this.ingester = new PullIngester(...);
} else {
    this.ingester = new BulkIngester(...);  // existing path
}

PullEngine Interface

public interface PullEngine {
    void initialize(int partitionCount);
    void write(int partition, String key, byte[] document);
    void flush();
    void shutdown();
}

• initialize(partitionCount) — creates or validates the topic/stream with the required number of partitions (one per primary shard).
• write(partition, key, document) — sends a document to a specific partition. The key is the document ID.
• flush() — flushes any buffered writes.
• shutdown() — closes the engine and releases resources.

Project Structure

A new Gradle project opensearch-pull-kafka under data-prepper-plugins/:

data-prepper-plugins/
├── opensearch/                          # existing
│   ├── src/main/java/.../
│   │   ├── OpenSearchSink.java
│   │   ├── Ingester.java
│   │   ├── BulkIngester.java
│   │   ├── PullIngester.java           # new
│   │   ├── PullEngine.java             # new interface
│   │   ├── PullIndexingConfig.java     # new config
│   │   └── DocumentFormatter.java      # new, shared format logic
├── opensearch-pull-kafka/               # new project
│   ├── build.gradle
│   ├── src/main/java/.../
│   │   └── KafkaPullEngine.java        # @DataPrepperPlugin implementing PullEngine
│   │   └── KafkaPullEngineConfig.java

I've started some refactoring to introduce the Ingester interface (#6795) specifically to support this second ingestion approach. With this refactoring we will use a new PullIngester to ingest data using pull-based ingestion.

The opensearch-pull-kafka project depends on the opensearch project (for the PullEngine interface) and kafka-plugins (for shared Kafka utilities). KafkaPullEngine uses KafkaCustomProducer from the existing kafka-plugins project internally for producing messages, reusing its authentication, serialization, and error handling.

Document Format

Pull-based ingestion expects documents in OpenSearch's default mapper envelope format:

{
  "_id": "abc123",
  "_version": 1714500000000,
  "_source": {"field": "value", "other": "data"},
  "_op_type": "index"
}

The PullIngester is responsible for constructing this envelope from Data Prepper events.
This is a key difference from BulkIngester, which builds BulkOperation objects for the Bulk API.

Field Mapping

Envelope Field	Source
_id	document_id expression (preferred) / document_id_field (deprecated) / auto-generated
_version	document_version expression / event timestamp millis / System.currentTimeMillis() fallback
_source	The event data (after SinkContext tag filtering)
_op_type	EventActionResolver.resolveAction() output

Supported Actions

Pull-based ingestion supports a subset of the actions available in the Bulk API:

Action	Supported	Notes
index	Yes	Default action. Writes the full document, replacing any existing version.
create	Yes	Writes only if the document does not already exist.
delete	Yes	Deletes the document by ID. _source may be omitted.
update	No	Pull-based ingestion does not support partial updates. Rejected at config validation.
upsert	No	Same — rejected at config validation.

If action or actions is configured with update or upsert and pull_indexing is enabled, the sink throws InvalidPluginConfigurationException during initialization.

Document ID

The document_id configuration option (a format expression like ${/doc_id}) is the preferred way to specify document IDs. The deprecated document_id_field (a plain field name) is also supported for backward compatibility. Both work the same way in pull-based ingestion as in bulk ingestion.

When neither is configured, the PullIngester must generate an ID because:

1. Pull-based ingestion provides at-least-once semantics — duplicate delivery is possible during shard recovery or Kafka rebalancing.
2. Without a stable document ID, duplicates cannot be deduplicated.
3. The ID determines which partition (and therefore which shard) receives the document.

Generation Algorithm

Use the same Flake ID algorithm as OpenSearch (UUIDs.base64UUID()): a 15-byte value composed of a sequence counter, timestamp, and machine identifier, Base64-encoded to 20 characters.
This ensures:

• IDs are compatible with what OpenSearch would generate internally
• Lucene-friendly ordering (the byte layout is optimized for Lucene's term dictionary)
• Cross-node uniqueness (via machine identifier component)

We will implement this in a DocumentIdGenerator utility class in the opensearch project, porting the logic from OpenSearch's TimeBasedUUIDGenerator.

Partition Assignment

The document ID determines the target partition. To ensure that a document lands on the correct shard, partition assignment must match OpenSearch's internal document routing:

partition = murmur3_hash(document_id) % number_of_partitions

This uses the same Murmur3 hash that OpenSearch uses in OperationRouting.generateShardId().

Routing is not supported in the initial implementation. The routing and routing_field configuration options are rejected when pull_indexing is enabled.

Note: If the partition_strategy: auto feature (OpenSearch RFC #21136) is enabled on the index, OpenSearch handles internal routing after ingestion. But for the default 1:1 shard-to-partition mapping, correct partition assignment from the producer side is required.

Versioning

Pull-based ingestion uses external versioning exclusively.
The version must be a long value. OpenSearch accepts a write only if new_version > current_version.

Using document_version

The existing document_version configuration option is a format expression (e.g., "${/timestamp}") that evaluates to a Long. This works directly with pull-based ingestion — the evaluated value becomes the _version field in the envelope.

When document_version is configured, the PullIngester evaluates it per event, same as BulkIngester does today.

Default Version

When document_version is not configured, the default version is the event timestamp as epoch milliseconds.

Rationale:

• Events naturally have timestamps, and "latest event wins" is the correct default for a streaming pipeline.
• Epoch millis fits the long type requirement.
• It is monotonically increasing under normal operation.
• OpenSearch's own field_mapping mapper documentation uses a timestamp field as the canonical version example.

If the event has no timestamp, System.currentTimeMillis() is used as a fallback.

document_version_type

When pull_indexing is enabled, the document_version_type configuration is ignored.
Pull-based ingestion always uses external versioning. If document_version_type is explicitly set to internal, the sink logs a warning that it will be ignored.

Future Consideration: Event __version Metadata

The RDS source recently added a document_version event metadata attribute (#6762) for denormalized document joins. The RDS StreamRecordConverter generates monotonic versions using timestamp_millis * 1000 + sequence_counter, ensuring strict ordering even when multiple events share the same millisecond. This metadata attribute is already set on events flowing through Data Prepper.

In a future iteration, PullIngester could check for this metadata attribute as a version source when document_version is not explicitly configured — preferring it over raw event timestamp millis since it provides stronger ordering guarantees. This would allow RDS-to-OpenSearch pipelines to use pull-based ingestion with correct version semantics out of the box. But this might require changes to OpenSearch itself.

Version Conflict Handling

When OpenSearch encounters a version conflict during pull-based ingestion, it silently drops the document (logged at debug level). This means:

• Out-of-order events with older timestamps are dropped, which is the desired behavior.
• If exact-once semantics are needed, users should configure explicit document_id and document_version with application-level versioning.

Index Creation

During PullIngester.initialize(), the sink creates the OpenSearch index with the settings required for pull-based ingestion. This includes the ingestion_source configuration that tells OpenSearch where to pull from.

The number of primary shards comes from the existing number_of_shards setting in the sink's index configuration (the same setting used for bulk ingestion). No new configuration is needed — the sink already knows the shard count because it creates the index.

// Pseudocode in PullIngester.initialize()
int shardCount = indexConfiguration.getNumberOfShards();

// Create the index with ingestion_source settings
createIndexWithIngestionSource(openSearchClient, indexName, shardCount, engineSettings);

// Initialize the engine (create/validate Kafka topic with one partition per shard)
engine.initialize(shardCount);

The index creation request includes:

PUT /my-index
{
  "settings": {
    "number_of_shards": 5,
    "index.replication.type": "SEGMENT",
    "ingestion_source": {
      "type": "kafka",
      "param.topic": "my-index-ingestion",
      "param.bootstrap_servers": "kafka-1:9092,kafka-2:9092",
      "pointer.init.reset": "earliest"
    }
  }
}

If the index already exists, initialization fails with a clear error message. This avoids the complexity of validating that an existing index has the correct ingestion source configuration, replication type, and shard-to-partition mapping. Users who need to manage the index lifecycle independently can do so outside of Data Prepper and use a future "attach to existing index" mode.

Backpressure

When the Kafka producer buffer is full, the PullIngester allows the Kafka producer's send() to block.
This causes output() to slow down, which holds records in the pipeline buffer longer, which in turn applies backpressure to the source. This is the standard backpressure model in Data Prepper — the buffer is the pressure boundary between source and sink, and a slow sink naturally causes the source to back off.

No special backpressure mechanism is needed in the PullIngester itself. The Kafka producer's buffer.memory and max.block.ms settings (configurable under producer) control how long the producer blocks before throwing an exception.

Implementation Phases

Phase 1: Multi-Partition Pull-Based Ingestion (MVP)

The goal is to wire up all components for pull-based ingestion with correct shard routing, targeting an existing OpenSearch index that already has ingestion_source configured. No default values
for fields that may change in later phases — all values are explicit.

• PullIndexingConfig configuration class using PluginModel for engine, marked @Experimental
• PullEngine interface in the opensearch project
• Configuration validation: reject update/upsert actions, dynamic index when pull_indexing is enabled
• PullIngester implementing Ingester with ingester selection logic in OpenSearchSink
• PullIngestionEnvelopeBuilder (event to envelope conversion: _id, _version, _source, _op_type)
• IndexRouter — determines target partition using Murmur3 hash of routing value matching OpenSearch's OperationRouting.generateShardId()
• IndexShardProvider — fetches shard count and ingestion topic name from the cluster's index settings (with Caffeine caching)
• DocumentIdResolver — resolves document ID from field or expression
• Partition count matches the number of primary shards (fetched from cluster)
• Topic name fetched from the index's ingestion_source.param.topic setting
• Routing value defaults to document ID; routing_field and routing expression are also supported
• New opensearch-pull-kafka Gradle project with KafkaPullEngine implementing PullEngine
• TopicManager — creates Kafka topic with correct partition count, expands partitions if topic already exists with fewer, waits for partition readiness
• KafkaPullEngine uses dependency injection via @Named and packagesToScan
• document_id is required — no auto-generation yet
• document_version is required — no default version logic yet
• action defaults to index (the only default in this phase)
• Shared retry mechanism via AbstractSink.doInitialize() handles transient cluster unavailability
• Metrics: documents written, documents failed
• Unit tests for all new components
• Plugin framework integration test (KafkaPluginEngineIT) verifying DI-based construction
• Integration test: produce documents through PullIngester, verify they appear in the OpenSearch index

Phase 2: Index Creation

• Data Prepper creates the OpenSearch index with ingestion_source settings during initialization
• Shard count from the sink's number_of_shards configuration
• Replication type set to SEGMENT as required by pull-based ingestion
• Index must not already exist — fail with a clear error if it does
• Topic name derived from index name when not explicitly configured

Phase 3: Document ID and Version Generation

• DocumentIdGenerator (Flake ID implementation matching OpenSearch's UUIDs.base64UUID())
• document_id becomes optional — auto-generated when not configured
• Version resolution: document_version expression or event timestamp millis default — document_version becomes optional
• DLQ integration for version evaluation errors

Phase 4: Production Readiness

• DLQ integration for all failure modes (serialization errors, Kafka produce failures)
• Enhanced metrics: partition distribution, envelope serialization errors
• Authentication support in KafkaPullEngineConfig (SASL, SSL)
• Producer tuning properties (compression_type, batch_size, linger_ms, buffer_memory)
• Performance benchmarking: pull vs. bulk ingestion throughput

Phase 5: Advanced Features (future)

• Dynamic index support (per-event index resolution)
• Attach to existing index mode (skip index creation, validate existing settings)
• Support for OpenSearch's field_mapping mapper as an alternative to the default envelope format
• Kinesis engine (opensearch-pull-kinesis)

Open Questions

1. Topic naming convention: When a topic name is not explicitly configured, should it default to the index name directly, or use a prefix/suffix (e.g., dp-pull-{index})?

Tasks