Inside the flight path of realtime ingestion in Apache Pinot (Part 3)

In Part 1 we followed the bytes. Kafka partitions, the consume loop, mutable segments, flush thresholds and the immutable build.

In Part 2 we followed the agreement. Servers report their offsets. The controller picks one committer and one winning offset. A single commit sequence publishes the segment to the whole cluster.

That protocol is correct. Every replica ends up with the same bytes. Only one commit path can ever reach COMMITTED. We ended that post feeling pretty good about ourselves.

Then someone asked what the partition consumer was doing while all of that happened.

Nothing.

That is the subject of this blog.

One note before we trace it. If you read the first two posts you already have the vocabulary and can skip this paragraph. If not, five names carry most of the weight here and all five are simpler than they sound. ZooKeeper, ZK for short, is the cluster’s notebook. It holds the facts every machine has to agree on, like which stream offsets belong to which segment. The IdealState is one page of that notebook. It records which server should be doing what with each segment, so rewriting that page is how Pinot publishes a decision. Helix is the messenger that reads the notebook and tells servers to act on it. A broker is the node a query actually lands on. It fans that query out to the servers holding the data and merges what comes back. And the deep store is durable shared storage, usually S3 or GCS, where finished segment files live.

Stuck on the taxiway

Trace the blocking flow one more time. Same steps as Part 2, but this time watch the clock instead of the state machine.

The consumer for a partition hits its flush threshold. That is the row count or size limit saying one segment has taken enough. It stops reading from Kafka. It sends the controller a segmentConsumed message saying where it stopped, then waits. The controller picks a winner. The winner is told to commit. Only now does the real work start. The server turns its in memory rows into an immutable segment on disk, writes every index, tars it up and uploads it to the deep store. When that finishes it reports back with segmentCommitEndWithMetadata to say the bytes are safe. Only then does the controller write down the offsets for the next segment and rewrite the IdealState to publish the change. Helix carries that out to the servers as an instruction to start consuming. That is the moment the partition consumer resumes.

Add it up. Your partition consumer is dark for the whole of build, upload, ZK writes and one Helix round trip.

Dark meaning what, though. Nothing breaks. Queries keep working. The committed segments are all still there. The mutable in memory copy is still being served right up until the swap. What you lose is freshness on one partition, which is exactly the thing you bought Pinot for.

And the bill is bigger than the pause. While you are dark, Kafka does not stop. Events pile up at the offset you walked away from, so when consumption resumes you are not caught up, you are behind on everything that arrived plus everything still arriving. Part 1 covered topic retention. If a pause ever outlasts it, you are not late any more, you are missing rows.

For a small segment all of that is a few seconds. Fine. Most tables never notice. If that is your table you can stop reading here and go enjoy your afternoon.

It stops being fine in three situations.

Big segments. Build time scales with rows and with how many indexes you asked for. A startree index needs a complete view of the segment data, so it can only be built once the segment is sealed and taking no more rows. Text and JSON indexes are not free either.

Shared build slots. Segment build on a server is bounded by a semaphore, so that a hundred partitions do not all try to build at once and blow up the cpu and heap. It also means partitions queue behind each other. StarTree’s 2024 year in review put it plainly. In rare cases of hitting concurrent segment build thresholds, the pause could stretch to tens of minutes.

Correlated flush boundaries. Partitions on the same topic tend to hit the row threshold at roughly the same moment, because they receive roughly the same traffic. So the pauses do not spread themselves out politely. They bunch up.

So just build faster?

That is where everyone goes first. It is worth taking seriously before we throw it out.

You can make the build cheaper. Drop the startree index, drop the text index, shrink the segment. Every one of those works. Every one costs you something you wanted. You added those indexes because queries needed them. Cutting them to speed up commits trades query latency for ingestion latency, which is not a trade. It moves the complaint to a different dashboard.

You can raise the build semaphore. Now a hundred partitions build at once and you are back to the cpu and heap problem the semaphore existed to prevent.

Look again at what the consumer is waiting for. Build. Upload. ZK writes. Helix round trip. All of it is real work that has to happen.

So the question is not how to make the work faster. It is why the consumer has to sit and watch it.

Clearance first, cargo later

Go back to the flow we just traced and sort the work by what it costs.

“Which offsets belong to this segment” is cheap. It is three small writes to ZooKeeper. Mark the old segment done, create metadata for the new one, flip the IdealState. Milliseconds.

“Here are the bytes for those offsets” is expensive. It is CPU to build every index, disk to write the tarball and a network upload to deep store. Seconds to minutes.

Now notice the order the blocking protocol runs them in. Expensive work first, cheap writes after it, the consumer blocked until both are done and Helix has delivered the transition.

So the consumer is waiting for the cheap work. The cheap work is waiting for the expensive work.

Ask why and there is no good answer. The controller knew the winning offset the moment it picked a committer. It has known which slice of the stream belongs to this segment since before the build started. It just does not tell anybody until the bytes land.

So stop doing that. Publish the decision, let the consumer go and publish the bytes when they are ready.

If that feels too easy, you are half right. The commit really is that easy. Nothing about the agreement changes. The controller still picks exactly one committer and one winning offset, exactly as in Part 2. All that changes is when the rest of the cluster hears about it. That part is a reordering. Reorderings are cheap.

The price is somewhere else. Worth naming now so it does not feel like a twist later. You are about to create a stretch of time where the cluster believes in a segment that has no bytes. Everything expensive about pauseless lives in that stretch. We will walk it properly further down.

First, a smaller problem. How do you even write down “we have agreed on this segment, but the bytes do not exist yet”?

A word for “agreed but not airborne”

Try it with the vocabulary we already have.

Every realtime segment carries a status in its ZK metadata. IN_PROGRESS means it is still consuming. DONE means it finished consuming and its bytes are committed to deep store. There is a third value, UPLOADED, but that is for segments an external job pushed in, so it does not help us here.

Leave our segment IN_PROGRESS and something will try to consume into it.

Mark it DONE and you break the thing that actually hurts. DONE is what the rest of Pinot reads as “the bytes are in deep store, here is the URL, go get them.” The rebalance job that moves segments between servers believes it. A replica that needs a copy believes it. Both go looking for a download URL nobody has written yet.

Two states cannot express three situations. So the enum grew a third value, COMMITTING, sitting between IN_PROGRESS and DONE. Consumed, agreed, offsets recorded, bytes not built. Any code that asks “can I download this segment?” gets a no while the segment is in flight.

COMMITTING only ever shows up on pauseless tables. A non pauseless table goes straight from IN_PROGRESS to DONE, exactly as before.

Final approach, resequenced

Same messages on the wire as Part 2. Different work behind them.

1. The server stops consuming and sends segmentConsumed.
2. The controller picks a winner and a winning offset, then replies COMMIT.
3. The winner calls segmentCommitStart.
4. The controller marks the metadata COMMITTING with the end offset, creates metadata for the next segment, flips the IdealState and then replies COMMIT_CONTINUE.
5. Helix tells the servers to start the next segment, the OFFLINE to CONSUMING transition. Ingestion resumes here.
6. The server builds the segment.
7. The server uploads to deep store.
8. The server calls segmentCommitEndWithMetadata.
9. The controller moves the metadata from COMMITTING to DONE and fills in the CRC, the size and the download URL.

Step 4 is the whole feature. In the blocking protocol those three ZK writes came after the build and the upload. Here they come before. Ingestion resumes four steps earlier and the expensive work carries on in the background.

Now, if you were reading step 4 closely, you have an objection.

Who is flying the segment with no bytes?

This is the question everyone asks. It is also the right question.

At step 4 the IdealState says the committing segment is ONLINE. Brokers will route queries to it. But the immutable bytes do not exist yet. Not on disk, not in deep store, not anywhere. So who answers?

The server that has them in memory.

The committer never dropped its mutable segment. It is still sitting there, still indexed, still serving queries, exactly as it was during consumption. The swap to the immutable copy happens later, when the build finishes. From the broker’s point of view nothing changed, because the broker was already routing to that server.

The other replicas converge the same way they did in Part 2, with one adjustment to what a non winner gets told.

• KEEP if the offset matches the winning offset. Build your own local copy. The metadata and the next consuming segment already exist, so you can start ingesting too.
• CATCH_UP if the offset is behind. Consume until you reach the winning offset.
• HOLD if the offset is ahead. The commit has not landed yet. You may be the next committer if the winner dies.

One detail is easy to miss. It bites in a fun way if you do. Pinot normally checks a locally built segment’s CRC checksum against the committed one in ZK. If they differ it downloads the committed copy instead of swapping. Sensible. Except during the COMMITTING window there is no committed CRC yet, so the check gets skipped for pauseless tables. Leave it in and a replica refuses its own perfectly good local build, then goes off to download bytes nobody has uploaded.

A replica might still genuinely need a copy during the window. Normally it would fetch one from deep store. During COMMITTING there is nothing in deep store to fetch, so Pinot falls back to peer download instead. The replica asks the other servers rather than the deep store, finds one that already has the segment online and pulls the bytes straight off it. Server to server, no durable copy involved.

That fallback is the reason pauseless with more than one replica refuses to start unless peerSegmentDownloadScheme is set. It is the setting that tells a server how to reach its peers. Without it the fallback has nowhere to go, so table config validation blocks the table up front rather than letting you discover it during an incident.

One thing is still unresolved. You probably spotted it several paragraphs ago. For the whole of that window, one server holds the only copy of those rows, in memory, with no tarball anywhere. What happens when it dies? Same window as before. We are nearly there.

One runway per partition

Releasing the consumer is not only the controller’s call. The server has a gate of its own. It is a good place to get this wrong.

Each partition has a ConsumerCoordinator holding a single permit. A consuming segment takes that permit for its whole life, which is how Pinot guarantees two segments of the same partition never consume at once. Now think about what happens if the committer holds that permit until its build finishes. The controller flips the IdealState, Helix delivers CONSUMING, the new segment starts up, reaches for the permit and blocks. All that work to move the ZK writes earlier. The consumer waits anyway.

So the permit is released at the start of the build rather than at the end.

Whether that is allowed at all is decided by ParallelSegmentConsumptionPolicy, which covers two windows. One is the build. The other is a download.

• ALLOW_ALWAYS consumes during both.
• ALLOW_DURING_BUILD_ONLY consumes during the build and waits during a download.
• ALLOW_DURING_DOWNLOAD_ONLY does the reverse.
• DISALLOW_ALWAYS waits for both.

Plain tables default to ALLOW_ALWAYS. Tables with dedup or partial upsert default to ALLOW_DURING_BUILD_ONLY when pauseless is on. That asymmetry looks arbitrary until you see why.

Dedup drops a row whose primary key it has seen before. Partial upsert merges a new row into the existing one for that key. Both keep a running map of keys in memory, so both need every earlier record reflected in that map before the next segment starts consuming. Get that wrong and you do not get a slow answer, you get a wrong one. A local build keeps that metadata in memory the whole time, so consuming alongside it is safe. A download replaces the segment from the outside. The metadata has to be reapplied first. So build is allowed and download is not.

Ordering is handled separately. The coordinator makes segment N+1 wait until segment N has been registered on that server, using the sequence number in the segment name. If that wait runs long, it re reads the IdealState to find out what the previous segment actually is, rather than assuming it is N.

Did the pause actually go away?

Fair question. The honest answer is that it depends on whether you check.

Pauseless does not make anything faster. The build takes as long as it took before and the upload costs what it cost before. All we did was stop the consumer from watching.

When the commit pause was the whole problem, that is enough. StarTree’s 2024 year in review put a number on it. One production deployment went from around 300 seconds of ingestion lag to 5 once consumers stopped waiting on commits. Same hardware, same tables, same stream. The only thing that changed is that nobody was watching the build any more.

It is not always the whole problem though. If some other gate in that chain is still closed, the wait is still there. It just moved somewhere less obvious, which is worse than where it started, because now it does not look like a commit pause.

There is one server metric that answers this, called prevSegmentWaitTimeMs. It measures how long the next segment sat waiting for the previous one to register. Close to zero means the consumer is genuinely not waiting. Anything else means the pause moved rather than disappeared. The ConsumerCoordinator is where to go looking.

Turbulence in the new window

Here is that window. This is the part that costs you something.

The blocking protocol published a segment in one controller call. Either it happened or it did not. Pauseless publishes it in two, separated by however long a build and an upload take. Seconds to tens of minutes.

Turn one atomic operation into two and you have created a window. A window can be interrupted. So walk it yourself before reading on. Two machines matter in there, the controller and the committer. Each can die on either side of its own work. That is most of the list.

The controller dies partway through commitStart. Those three ZK writes in step 4 are not one transaction, so you can land in the middle. Metadata marked COMMITTING, IdealState still saying CONSUMING. Nobody finishes the job on its own, because the server already got its answer and moved on to building. The controller runs a periodic validation sweep that looks for exactly this shape and finishes the job. It creates the next segment metadata and flips the IdealState. For a pauseless table the expected status after that first controller call is COMMITTING rather than DONE. The repair logic keys off that difference.

The committer dies during the build. Now the segment sits in COMMITTING and the bytes are never coming, because the only machine that had them in memory is gone. If every replica for that segment is in ERROR, the controller picks a live server and triggers reingestion through the server’s /reingestSegment endpoint. That server replays the exact offset range recorded in the ZK metadata, builds the segment and uploads it. This only works because the end offset was written at commitStart. The decision we published early is what makes the replay deterministic.

There is a guard on it. Reingestion for tables with dedup or partial upsert only runs when the table’s disasterRecoveryMode allows it, because replaying records through a stateful path can produce wrong data rather than merely late data. If the mode says no, the controller counts the segment under pauselessSegmentsInUnrecoverableErrorCount and leaves it alone. That is a deliberate choice. It is the right one. A loud stuck segment beats a quiet wrong one.

The upload never lands in deep store. A segment whose bytes live on exactly one server is a segment you are going to lose eventually. So for pauseless tables that same sweep checks for missing deep store copies on every pass, not only when deep store retry is switched on.

Nobody knows what is in flight. Every one of the repairs above starts with knowing which segments are mid window. The controller keeps a list of in flight segments in ZooKeeper, added at commitStart and removed at commitEnd. The sweep rebuilds that list from actual segment metadata on every pass, so a controller restart does not lose the thread. Rebalance reads it too, because moving a segment that has no bytes yet is a different problem from moving a normal one.

So that is where the complexity went. Not into the commit path, which barely changed at all, but into sweep and repair. You cannot make two ZK calls atomic across a multi minute build, so instead you make every partial state recognisable and give the sweep a way to finish the job. Most of the code pauseless added is that plan.

Preflight checks

Turning it on is two settings in the table config. Add pauselessConsumptionEnabled next to your existing stream configs and give validationConfig a peerSegmentDownloadScheme.

{
  "ingestionConfig": {
    "streamIngestionConfig": {
      "streamConfigMaps": [ { "...": "..." } ],
      "pauselessConsumptionEnabled": true
    }
  },
  "validationConfig": {
    "peerSegmentDownloadScheme": "http"
  }
}
Code language: JSON / JSON with Comments (json)

That second setting is not optional decoration. The config validator will tell you so.

With replication greater than 1, peerSegmentDownloadScheme must be set and valid. Without it, a replica can get stuck waiting on a deep store copy that has not been written yet.

Replication 1 is allowed, but warns. There is no peer to fall back on, so a single server failure during the COMMITTING window means re-ingestion or nothing.

Four metrics worth wiring up before you flip the switch, on top of prevSegmentWaitTimeMs from earlier.

• pauselessConsumptionEnabled (controller and server) confirms the table is genuinely on the pauseless path, which is worth checking rather than assuming.

• pauselessSegmentsInErrorCount (controller) counts segments in ERROR on a pauseless table.

• pauselessSegmentsInUnrecoverableErrorCount (controller) counts segments the controller has decided not to auto-repair. Page on this one.

• segmentMissingDeepStoreLink (controller) means a commit finished without a deep store location.

Mission accomplished

Pauseless consumption does not make the commit faster. The build takes as long as it takes, the upload costs what it costs. Nothing here changed either. What it does is stop your consumer standing around watching.

The trade is worth stating plainly. You buy a shorter pause with a more complicated recovery story. A segment can now exist as an agreement before it exists as bytes. The system needs a plan for every way that window can end badly.

So the decision is not really “is pauseless better.” It is whether freshness is the thing your product sells. If it is, turn it on and watch prevSegmentWaitTimeMs. If your segments build in two seconds and nobody has ever filed a ticket about lag, you are buying complexity you do not need.

And that is the flight path, end to end. Part 1 followed the bytes, from an offset in a Kafka partition to an indexed segment on disk. Part 2 followed the agreement, how three replicas end up believing exactly the same thing about exactly the same slice of the stream. This post followed the clock. It costs something to keep that agreement without making the stream wait for it.

So that is the whole of it. Three segment states, two controller calls, the window between them and a sweep that tidies up when something dies inside it. Enough to open the code and know what you are looking at. Enough to decide whether your own tables want any of it.

Thanks for flying along.