Pauseless Ingestion #2: Handle Failure scenarios without DR #14798

9aman · 2025-01-13T09:46:46Z

Pauseless Ingestion Failure Resolution

Please refer to PR: #14741 for happy path. This PR aims to only cover the failure scenarios. Once the above one is merged a better diff covering only failures will be visible.

To view only diff covering failure scenarios, for the time being, refer to:

Summary

This PR aims to provide ways to resolve the failure scenarios that we can encounter during pauseless ingestion. The detailed list of failure scenarios can be found here: link along with the failure handling strategies: link

Following sequence diagrams summarizes the failure scenarios and the resolution.

Failure Scenarios & Resolution Approaches

Failures encountered during the commit protocol can be categorized into two types: recoverable and unrecoverable failures.

Recoverable failures are those in which at least one of the servers retains the segment on disk.

Unrecoverable failures occur when none of the servers have the segment on disk.

Recoverable Failures

Recoverable failures will be addressed through RealtimeSegmentValidationManager. This approach will handle scenarios such as upload failures and incomplete commit protocol executions.

The controller or server can run into issues in between any of the steps of the commit protocol as listed below:

Request Type: COMMIT_START

Update the Segment ZK metadata for the committing segment (seg__0__0)
- Change status to COMMITTING
- Set endOffset
Create Segment ZK metadata for the new segment (seg__0__1) with status IN_PROGRESS
Update the Ideal State for the:
- Committing segment (seg__0__0) to ONLINE
- New/ Consuming segment (seg__0__1) to CONSUMING

Request Type: COMMIT_END_METADATA
4. Update Segment ZK metadata for the committing segment (seg__0__0):
- Change status to DONE.
- Update deepstore url.
- Any additional metadata.

The RealtimeSegmentValidationManager figures out which step of the commit protocol failed and how can it be fixed. This is very similar to how commit protocol failures were handled before with some minor changes.

Non-recoverable Failures

These failures require ingesting the segment again from upstream, followed by build, upload and ZK metadata update.

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

...r/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java

KKcorps · 2025-01-14T02:08:01Z

pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java

-      return;
+    TableConfig tableConfig = indexLoadingConfig.getTableConfig();
+    // For pauseless tables, we should replace the segment if download url is missing even if crc is same
+    if (!PauselessConsumptionUtils.isPauselessEnabled(tableConfig)) {


Should also LOG that since pauseless is enabled we are not going directly for segment download but waiting for catchup

@KKcorps I don't understand why this is different for pauseless. I see you have added these along with reingestion tests.

.../org/apache/pinot/integration/tests/realtime/FailureInjectingRealtimeSegmentDataManager.java

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

KKcorps · 2025-01-14T02:12:11Z

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

+  private static final ConcurrentHashMap<String, AtomicBoolean> SEGMENT_INGESTION_MAP = new ConcurrentHashMap<>();
+
+  // Semaphore to enforce global concurrency limit
+  private static final Semaphore REINGESTION_SEMAPHORE = new Semaphore(MAX_PARALLEL_REINGESTIONS);


Make this count configurable

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

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

…jor fields of SegmentZKMetadata

9aman · 2025-01-24T07:23:08Z

pinot-common/src/main/java/org/apache/pinot/common/metadata/segment/SimpleSegmentMetadata.java

+ * SimpleSegmentMetadata deserialized = objectMapper.readValue(json, SimpleSegmentMetadata.class);
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class SimpleSegmentMetadata {


@Jackie-Jiang have added this to simplify ser/deser of SegmentZkMetadata. Added few getters for easy access from simpleFields for important keys like crc.

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

Jackie-Jiang · 2025-01-28T01:23:41Z

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

+    ImmutableSegmentDataManager immutableSegmentDataManager = (ImmutableSegmentDataManager) segmentDataManager;
+    SegmentMetadataImpl segmentMetadata =
+        (SegmentMetadataImpl) immutableSegmentDataManager.getSegment().getSegmentMetadata();
+    SegmentZKMetadata segmentZKMetadata = getSegmentZKMetadata(segmentMetadata);


We should be able to reuse the methods in ZKMetadataUtils. You can move ZKMetadataUtils into pinot-common

Jackie-Jiang · 2025-01-28T01:28:43Z

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

+      // 1. DONE for normal consumption
+      // 2. COMMITTING for pauseless consumption
+      Status statusPostSegmentMetadataUpdate =
+          PauselessConsumptionUtils.isPauselessEnabled(tableConfig) ? Status.COMMITTING : Status.DONE;


Check this upfront, instead of on each partition

Jackie-Jiang · 2025-01-28T01:37:44Z

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

+
+    // Define upload methods in order of preference
+    List<UploadAttempt> uploadAttempts = Arrays.asList(
+        // Primary method


If the primary method failed, we should abort the committing segment fix because fallbacks won't be able to handle that

This has been added for backward compatibility.
I initially relied only on the primary function but that introduced Backward compatibility issues.

Jackie-Jiang · 2025-01-28T01:39:27Z

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

+    if (segmentZKMetadata.getStatus() == Status.COMMITTING) {
+      LOGGER.info("Updating additional metadata in ZK for segment {} as pauseless is enabled",
+          segmentZKMetadata.getSegmentName());
+      segmentZKMetadata.setStartTime(uploadedSegmentZKMetadata.getStartTimeMs());


Should we simply put all simple fields from the uploaded ZK metadata into current?

Have update the function to copy all the simpleFields. The SegementZkMetadata returned by the uploadLLSegmentToDeepStore now also the previously missing fields like creationTime and numReplicas

… segment API's

Jackie-Jiang

The new added tests are quite slow. PauselessRealtimeIngestionCommitEndMetadataFailureTest itself is taking more than 5 minutes, and we are adding 4 similar tests. Are we able to merge them into one and reduce the overall testing time?

…can only be updated after a fixed time has elapsed Reduced the time requirements by creating a FakePauselessLLCRealtimeSegmentManager.

9aman · 2025-01-29T06:31:16Z

The new added tests are quite slow. PauselessRealtimeIngestionCommitEndMetadataFailureTest itself is taking more than 5 minutes, and we are adding 4 similar tests. Are we able to merge them into one and reduce the overall testing time?

@Jackie-Jiang I have updated the tests. Each test now runs in under a minute.

…ache#14798)

9aman and others added 15 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

This was referenced Jan 13, 2025

Add DR and Failure scenario handling for Pauseless Ingestion #14794

Closed

Pauseless ingestion without failure scenarios #14741

Merged

KKcorps requested a review from Jackie-Jiang January 14, 2025 01:59

Jackie-Jiang added feature release-notes Referenced by PRs that need attention when compiling the next release notes ingestion real-time labels Jan 14, 2025