[PEP] stream plugin that supports object-storage micro-batches with optional inline batch support

[udaysagar2177]

Summary:

Introduce a new stream plugin that can handle external files or inline batches for real-time ingestion.

Motivation:

• Reduce Kafka cross-AZ replication, production, and consumption network costs.
• Kafka delivers sub-second message latency but requires substantial infrastructure; many use cases can tolerate higher latency at lower cost.
• High-volume traffic can exceed Kafka broker throughput or retention limits, necessitating complex operational management.

Core Mechanics

• Partition-level consumers receive micro-batch descriptor records instead of individual events.
• Each record triggers a controlled background fetch logic that downloads the referenced object via PinotFS.
• A dedicated thread extracts events using the configured RecordReader and pushes them into a bounded in-memory queue.
• The LLC consumer remains unchanged and reads from this in-memory queue as if it were a normal streaming source.

Offset Model

• Offsets use a serialized JSON structure similar to the Kinesis stream plugin.
• JSON tracks:
  • The Kafka record offset carrying the micro-batch descriptor record.
  • The intra-file (or intra-batch) event offset.
• Supports replay correctness, restart recovery, and stable start/end offset behavior.

Other advantages:

• Enables real-time ingestion of Avro or Parquet files without complicating the architecture with Spark or a separate batch ingestion job.
• Enables improved compression efficiency using inline batches compared to the message-per-event model.

Micro-batch descriptor protocol

• The micro-batch protocol defines deterministic sub-selection rules.
• Consumers may extract:
  • The full batch, or
  • Only the assigned-partition subset,
• Replay semantics remain fully stable.

Expected Outcome

A stream plugin that supports object-storage micro-batches with optional inline batch support, reduces the impact of Kafka, and simplifies ingestion pipelines for file-based formats.

If this proposal aligns with the project’s direction. I would be happy to move it forward and submit a pull request for review.

[Jackie-Jiang]

Another way to support micro-batches is to use offline table + a minion task to periodically merge small segments into proper sized ones. How often will the micro-batch being created?

[udaysagar2177]

The micro-batch ingestion interval can be as low as 5 to 10 seconds. Such low interval is mainly for durability and high data volume reasons. The idea is to reduce Kafka costs and complexity while still achieving near-real-time ingestion with minimal overhead.

Using a minion task to merge many small segments at such a frequency seems feasible, but it requires upstream processes to generate those small segments whether via a Spark job converting Avro files or the application itself. Both approaches introduce non-trivial work, and there’s a risk that segments could accumulate faster than the minions can merge them.

With a real-time micro-batch ingestion approach, we could still leverage Kafka for descriptor records while producing larger segments, similar to the “one Kafka message per event” model. During periods of higher load, micro-batches reside on durable storage until the pipeline scales and Kafka topic artitions are re-assigned.

[kishoreg]

+100 neat idea. I like it

[udaysagar2177]

Thanks! Please let me know if you’d prefer a draft design doc, or whether proceeding directly to a PR is okay.

[Jackie-Jiang]

A design doc stating the core part (e.g. file descriptor format, high level data flow) would help reviewers understand the code. I'd suggest creating one along with the PR.

[mayankshriv]

+1 neat idea.

[ankitsultana]

@udaysagar2177 : have you folks evaluated Warpstream? With the proposed approach I think we'll end up implementing part of what Warpstream's consumer does in Pinot. (hasty generalization but I suppose you get the idea)

Also, I haven't taken a deeper look yet, but I suppose there must be a way to do this already with the abstractions we already have? For instance, the offset is just a number which could be a key in some database that each vendor can control completely.

Similarly, Kafka and stream consumers already do micro batching so instead of fetching a batch of messages from kafka one can simply fetch it from object store.

[udaysagar2177]

Thanks for bringing up WarpStream. That’s a helpful comparison, and I agree there’s some conceptual overlap. I haven’t done a deep evaluation of WarpStream, but a few thoughts of my own suggest exploring this approach within the existing Pinot/Kafka abstractions.

1. Ecosystem and experimentation considerations
   WarpStream seems valuable for reducing Kafka-related costs, but being closed source makes prototyping and adoption a bit harder. Depending on usage and pricing, introducing WarpStream could add incremental costs. There is also an open-source alternative called AutoMQ, though adopting it without paid support would require operating an additional system.

2. Kafka as a lightweight coordination layer
   Kafka-based pipelines are likely to exist for other use cases in a system architecture. In that context, maintaining a small Kafka topic for micro-batch descriptors appears manageable. Kafka benefits from strong community support and availability through managed service offerings.

3. File-centric ingestion as a first-class path
   In some pipelines, data is naturally produced as Avro or Parquet in object storage to serve other downstream consumers. For those cases, creating a Kafka-style message production model solely for Pinot ingestion may be less desirable. Leveraging the PinotFS abstraction allows ingestion to remain closer to existing file-based workflows without requiring additional frameworks (e.g., Spark or Minion) to transform file data into a Kafka stream.

Caveats and considerations

Data routing generally needs to be determined before file generation so files can map cleanly to Kafka partitions consumed by Pinot. If that isn’t the case, files may need to be split or reorganized, which could increase object-store API costs and system complexity. To address this:

• Referencing exact byte ranges within a file is possible, but this approach would diverge from widely adopted file formats and their tooling ecosystem.
• Embedding filtering logic in the micro-batch descriptor is another option, though it could introduce additional design complexity.

The current abstractions do support micro-batch processing, and I am actively working through the necessary implementation details. While it appears feasible, I see a few areas that might complicate it, such as handling consumer poll intervals to accommodate rebalances during micro-batch processing, reduced visibility into backlog, increased compute load to parse data for Pinot servers, etc. I’m happy to submit a work-in-progress PR to help us assess whether this is still a good idea.