Pauseless Consumption #3: Disaster Recovery with Reingestion #14920

KKcorps · 2025-01-27T04:21:17Z

This PR adds support for Disaster recovery for Pauseless Ingestion along with Reingestion. These changes help solve the scenario where real-time segments permanently fail to transition out of ERROR state, leading to data gaps. With reingestion, Pinot can recover such segments, ensuring availability and correctness of real-time data.

During Pauseless ingestion, a ONLINE segment can wind up in an ERROR state if its commit fails due to server restart and there are no other replicas. Currently in pinot, there is no way to recover from such failures.

Reingestion Flow

Segments that fail to commit or end up in ERROR state can now be re-ingested by calling a new endpoint (/reingestSegment) on the server.

The ReIngestionResource reconstructs the segment from the stream, builds it, and commits it, ensuring that offline peers and the deep store get updated properly.

If successful, the re-ingested segment transitions from ERROR to ONLINE.

New APIs introduced:

Get Running Re-ingestion Jobs (Server)

GET /reingestSegment/jobs

Returns all currently running re-ingestion jobs with their status information.

Response

Type: JSON
Contains array of jobs with:
- jobId: Unique identifier
- tableNameWithType: Table being processed
- segmentName: Segment being re-ingested
- startTimeMs: Job start timestamp

Responses

200: Success - List of running jobs

Re-ingest Segment (Server)

POST /reingestSegment

Asynchronously re-ingests a segment with updated configurations.

Request Body

Type: JSON
Required fields:
- tableNameWithType: Table name with type (e.g. "myTable_REALTIME")
- segmentName: Name of segment to re-ingest

Response

Type: JSON
Contains:
- jobId: Unique identifier for tracking progress
- message: Success confirmation

Responses

200: Success - Job started successfully
429: Too Many Requests - Parallel job limit reached
404: Not Found - Table/segment not found
409: Conflict - Segment already being re-ingested
500: Internal Server Error - Server initialization issues

Complete Re-ingest Segment (Controller

POST /reingestSegment

Uploads the segment to deep store, updates ZK metadata to DONE and resets the segment

Same signature as /v2/segment upload with multi-part form data except that it doesn't allow refresh

Reingestion data flow

sequenceDiagram
    participant Controller
    participant Server
    participant ReIngestionResource
    participant SimpleRealtimeSegmentDataManager

    Controller->>Controller: Finds ERROR segment in validation task
    Controller->>Controller: Pick one alive server from IS for segment
    Controller->>Server: POST /reIngestSegment (tableName, segmentName)
    Server->>ReIngestionResource: ReIngestionRequest
    ReIngestionResource->>SimpleRealtimeSegmentDataManager: startConsumption()
    SimpleRealtimeSegmentDataManager-->>SimpleRealtimeSegmentDataManager: Consume data & Build Segment
    SimpleRealtimeSegmentDataManager->>ReIngestionResource: Segment Tar File
    ReIngestionResource->>Controller: POST (/segment/completeReingestion)
    Server-->>Controller: 200 OK (Reingestion complete)
    Controller->>Controller: Copy segment to deep store
    Controller->>Controller: Update ZK segment status to DONE
    Controller->>Server: Reset segment to ONLINE

Reingestion design diagram

flowchart TD
    A[Start reingestion request] --> B[Check concurrency <br> & segment ingestion map]
    B --> C{Already ingesting?}
    C -- Yes --> D[Return 409 conflict]
    C -- No --> E[Acquire semaphore, set ingesting]
    E --> F[Create & start SimpleRealtimeSegmentDataManager]
    F --> G{Consume from<br>startOffset to endOffset}
    G --> H[Build & tar segment]
    H --> I[Push metadata to controller]
    I --> J[Trigger reingestion completion API]
    J --> K[Release semaphore, mark not ingesting]
   K --> L[Done]

1. Changing FSM 2. Changing the 3 steps performed during the commit protocol to update ZK and Ideal state

1. Changes in the commit protocol to start segment commit before the build 2. Changes in the BaseTableDataManager to ensure that the locally built segment is replaced by a downloaded one only when the CRC is present in the ZK Metadata 3. Changes in the download segment method to allow waited download in case of pauseless consumption

…segment commit end metadata call Refactoing code for redability

… ingestion by moving it out of streamConfigMap

…auseless ingestion in RealtimeSegmentValidationManager

…d by RealtimeSegmentValitdationManager to fix commit protocol failures

…g commit protocol

… checks

…ption is enabled or not

…eepstore path with fallbacks

KKcorps · 2025-02-04T16:23:20Z

When a segment is re-ingested, it should be DONE instead of UPLOADED. I didn't find the semaphore related logic. Is the PR description up to date?

During reingestion segment metadata upload is triggered after segment building. Hence, the segment status appears as "UPLOADED" rather than "DONE".

Upon upload completion, a segment refresh operation is triggered, which replaces the failed RealtimeSegmentDataManager with an ImmutableSegmentDataManager. During this replacement process, the system automatically releases associated semaphores.

Once the segment is refreshed, we send a reset message so that it starts showing up as ONLINE in EV

pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java

...ava/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java

pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java

Jackie-Jiang · 2025-02-06T01:09:22Z

pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java

      @Nullable String sourceDownloadURIStr, String segmentDownloadURIStr, @Nullable String crypterName,
      long segmentSizeInBytes, boolean enableParallelPushProtection, boolean allowRefresh, HttpHeaders headers)
      throws Exception {
+    completeSegmentOperations(tableNameWithType, segmentMetadata, uploadType, finalSegmentLocationURI, segmentFile,


Let's not coupling the logic of reingestion into current regular segment complete handling. Currently the logic is very hard to read, and we are coupling reset with reingestion, which is not correct.
We can add a new method completeReingestedSegmentOperations(). Reingested segment holds different assumptions then regular uploaded segment.

...ain/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java

...main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java

pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReIngestionResource.java

Jackie-Jiang · 2025-02-06T01:49:56Z

pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReIngestionResource.java

+      try {
+        int partitionGroupId = llcSegmentName.getPartitionGroupId();
+
+        Map<String, String> streamConfigMap = IngestionConfigUtils.getStreamConfigMaps(tableConfig).get(0);


Is there a way to know which stream config to use to generate the segment?

Not possible. Ideally though in multi-stream only broker and topic names should differ across stream configs, rest everything should remain the same

pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReIngestionResource.java

…pache#14920)

lnbest0707-uber · 2025-02-25T23:49:31Z

This is a great feature. And I believe we could expand its usage beyond failure recovery.
E.g. we could combine the feature with this proposal #14815 for better ingestion freshness guarantee.

Have a similar ingestAdhoc endpoint to, instead of re-ingesting, ingest a segment based on the input partition and offset.
Hooked together with the proposal in Pinot real-time data ingestion freshness guarantee #14815. The controller skips the offset based on freshness status, and also spawns the asynchronous ingestAdhoc job.

lnbest0707-uber · 2025-02-26T19:31:21Z

...main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java

+    LLCSegmentName llcSegmentName = new LLCSegmentName(segmentZKMetadata.getSegmentName());
+
+    _segmentName = segmentZKMetadata.getSegmentName();
+    _partitionGroupId = llcSegmentName.getPartitionGroupId();


This will be incompatible with multi-stream ingestion. We need to use the streamPartitionGroupId to do the ingestion. Refer to the RealtimeSegmentDataManager

lnbest0707-uber · 2025-02-26T19:37:52Z

...main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java

+
+    _logger = LoggerFactory.getLogger(StatelessRealtimeSegmentWriter.class.getName() + "_" + _segmentName);
+
+    Map<String, String> streamConfigMap = IngestionConfigUtils.getStreamConfigMaps(_tableConfig).get(0);


9aman and others added 30 commits January 2, 2025 16:57

Controller side changes to allow pauseless ingestion:

6defce3

1. Changing FSM 2. Changing the 3 steps performed during the commit protocol to update ZK and Ideal state

Minor improvements to improve readability

54ab7b3

Fixes in the commit protocol for pauseless ingestion to complete the …

012da87

…segment commit end metadata call Refactoing code for redability

Changing the way to enable pausless post introduction of multi-stream…

a97847f

… ingestion by moving it out of streamConfigMap

WIP: Changes in the expected state while performing recovery during p…

2c2ba86

…auseless ingestion in RealtimeSegmentValidationManager

Adding changes for ensurePartitionsConsuming function that is utilize…

4d7c893

…d by RealtimeSegmentValitdationManager to fix commit protocol failures

Add Server side Reingestion API

a041a75

run segment level validation

b6d0904

Add method to trigger reingestion

58f6c51

Linting fixes

d6313b3

Adding integration tests for 3 failure scenarios that can occur durin…

ca6134a

…g commit protocol

WIP: Reingestion test

845f616

Fix bug with indexes in reingestion

d2dd313

Add tests for reingestion

fb34fc8

Fix controller url in Reingestion

50725bd

Support https and auth in reingestion

e74d360

Merge branch 'master' into resolve-failures-pauseless-ingestion

ce3c851

Removing checks on default crc and replacing them with segment status…

d6208a6

… checks

Formatting improvements

7f5b720

Allowing null table config to be passed for checking pauseless consum…

3f05b2f

…ption is enabled or not

Ensuring upload retries are backward compatible

2012e38

Removing unnecessary code

7b9da37

Fixing existing test cases and adding unit tests to check upload to d…

ded8962

…eepstore path with fallbacks

Refactor test cases to reduce repetition

e703d84

Add missing header file

c2fda4a

Fix reingestion test

c836009

refactoring file upload download client

7974ab9

Removing pauselessConsumptionEnabled from index config

c08f841

Remove reingestion code

c4b99bd

KKCorps added 3 commits February 4, 2025 20:15

Refactoring

d035249

refaactoring

bc8a65b

Honour segment build semaphore during reingestion

f082d24

KKCorps added 2 commits February 5, 2025 07:22

Fix test

d1ad30b

Add a seperate API to upload reingested segments

7e47dd5

KKcorps force-pushed the pauseless-reingestion branch from 7432261 to fd79338 Compare February 5, 2025 13:48

Cleanup code

5a42c28

KKcorps force-pushed the pauseless-reingestion branch from fd79338 to 5a42c28 Compare February 5, 2025 14:09

Jackie-Jiang reviewed Feb 6, 2025

View reviewed changes

KKCorps and others added 10 commits February 6, 2025 08:57

Replace reIngest with reingest

1551685

Replace reIngest with reingest

837aa26

Ensure correctness in reingestion

09e8583

Decouple reingest completion and upload ZK path

3d6fdf4

Refactor Stateless segment writer to have least number of arguments

b525c7e

Fix segment reset for pauseless tables

f7ae25f

Cleaning up Stateless segment writer

3c17957

Rename metric

a3fa25c

Misc cleanup and logic decoupling

61ebb23

nit

62316c0

Jackie-Jiang approved these changes Feb 12, 2025

View reviewed changes

Jackie-Jiang merged commit 7fcb82c into apache:master Feb 12, 2025
20 of 21 checks passed

zeronerdzerogeekzerocool pushed a commit to zeronerdzerogeekzerocool/pinot that referenced this pull request Feb 20, 2025

Pauseless Consumption apache#3: Disaster Recovery with Reingestion (a…

4eda495

…pache#14920)

lnbest0707-uber reviewed Feb 26, 2025

View reviewed changes

lnbest0707-uber mentioned this pull request Mar 19, 2025

Retry on segment cannot build #15234

Merged

This was referenced Mar 19, 2025

Pauseless Consumption #10147

Closed

Allows segments deletion in build for pauseless tables #15299

Merged

lnbest0707-uber mentioned this pull request Jul 29, 2025

Pauseless consumption is not compatible with multi-topic ingestion #16451

Open


		_logger = LoggerFactory.getLogger(StatelessRealtimeSegmentWriter.class.getName() + "_" + _segmentName);

		Map<String, String> streamConfigMap = IngestionConfigUtils.getStreamConfigMaps(_tableConfig).get(0);

Pauseless Consumption #3: Disaster Recovery with Reingestion #14920

Pauseless Consumption #3: Disaster Recovery with Reingestion #14920

Uh oh!

Conversation

KKcorps commented Jan 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Reingestion Flow

New APIs introduced:

Get Running Re-ingestion Jobs (Server)

Re-ingest Segment (Server)

Complete Re-ingest Segment (Controller

Reingestion data flow

Reingestion design diagram

Uh oh!

KKcorps commented Feb 4, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Jackie-Jiang Feb 6, 2025

Choose a reason for hiding this comment

Uh oh!

KKcorps Feb 6, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Jackie-Jiang Feb 6, 2025

Choose a reason for hiding this comment

Uh oh!

KKcorps Feb 6, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lnbest0707-uber commented Feb 25, 2025

Uh oh!

lnbest0707-uber Feb 26, 2025

Choose a reason for hiding this comment

Uh oh!

lnbest0707-uber Feb 26, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

KKcorps commented Jan 27, 2025 •

edited

Loading