[Pipe] Resume TsFile parsing after retryable failures#18261
Open
Caideyipi wants to merge 1 commit into
Open
Conversation
jt2594838
reviewed
Jul 22, 2026
Comment on lines
+793
to
+794
| // Reacquire parser memory after a previous failure yielded the active parser slot. | ||
| waitForResourceEnough4Parsing((long) ((1 + Math.random()) * 20 * 1000)); |
Contributor
There was a problem hiding this comment.
Backoff does not work on this?
Comment on lines
+861
to
+865
| final int shift = Math.min(30, Math.max(0, retryCount - 1)); | ||
| if (initialBackoffInMs > (maxBackoffInMs >> shift)) { | ||
| return maxBackoffInMs; | ||
| } | ||
| return Math.min(maxBackoffInMs, initialBackoffInMs << shift); |
Contributor
There was a problem hiding this comment.
What does Math.min(30 mean
Comment on lines
+175
to
193
| final PipeTsFileInsertionEvent tsFileInsertionEvent = (PipeTsFileInsertionEvent) event; | ||
| tsFileInsertionEvent.consumeTabletInsertionEventsWithRetry( | ||
| event1 -> { | ||
| try { | ||
| pipeProcessor.process(event1, outputEventCollector); | ||
| } catch (PipeRuntimeOutOfMemoryCriticalException e) { | ||
| throw e; | ||
| } catch (Exception e) { | ||
| throw new PipeException(e.getMessage(), e); | ||
| } | ||
| }, | ||
| "PipeProcessorSubtask::executeOnce"); | ||
| tsFileInsertionEvent.close(); | ||
| if (tsFileInsertionEvent.isGeneratedByHistoricalExtractor()) { | ||
| PipeTerminateEvent.markHistoricalTsFileSplit( | ||
| tsFileInsertionEvent.getPipeName(), | ||
| tsFileInsertionEvent.getCreationTime(), | ||
| regionId); | ||
| } |
Contributor
There was a problem hiding this comment.
Double-check the risk of resource leak after this modification.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
This is the parser progress recovery part of the Pipe resilience work. It is independent of #18260 and uses the existing parser memory reservation API on master.
Preserve parsing progress
PipeTsFileInsertionEventnow retains:If tablet consumption fails, the event yields its active parser memory reservation but keeps the parser, iterator, and pending tablet. The next subtask retry reacquires parser memory and resumes from the exact same tablet object instead of reopening and reparsing the TsFile from the beginning.
Parser lifecycle
Call sites that split Pipe TsFile events now close the parser only after successful completion. Processing exceptions stop at the failed tablet and propagate immediately. Non-Pipe TsFile implementations retain their original finally-close behavior.
Explicit event close and resource finalization still release the pending tablet and parser resources.
OOM retry backoff
Local parser OOM retries use bounded exponential backoff derived from the existing Pipe memory retry interval and maximum retry count. When the retry policy decides to yield, later subtask retries resume from saved progress.
Tests
TsFileInsertionEventParserTestverifies that an OOM does not close the parser and that the next retry receives the identical failed tablet object. The full parser test class passes.This PR has:
Key changed/added classes
PipeTsFileInsertionEventPipeProcessorSubtaskPipeEventCollectorAggregateProcessorDownSamplingProcessorWebSocketSinkTsFileInsertionEventParserTest