Skip to content

[Pipe] Resume TsFile parsing after retryable failures #18261

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[Pipe] Resume TsFile parsing after retryable failures #18261

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 6 additions & 9 deletions ...node/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,15 +147,12 @@ private void parseAndCollectEvent(final PipeTsFileInsertionEvent sourceEvent) th
return;
}

try {
sourceEvent.consumeTabletInsertionEventsWithRetry(
this::collectParsedRawTableEvent, "PipeEventCollector::parseAndCollectEvent");
if (sourceEvent.isGeneratedByHistoricalExtractor()) {
PipeTerminateEvent.markHistoricalTsFileSplit(
sourceEvent.getPipeName(), sourceEvent.getCreationTime(), regionId);
}
} finally {
sourceEvent.close();
sourceEvent.consumeTabletInsertionEventsWithRetry(
this::collectParsedRawTableEvent, "PipeEventCollector::parseAndCollectEvent");
sourceEvent.close();
if (sourceEvent.isGeneratedByHistoricalExtractor()) {
PipeTerminateEvent.markHistoricalTsFileSplit(
sourceEvent.getPipeName(), sourceEvent.getCreationTime(), regionId);
}
}

Expand Down
41 changes: 18 additions & 23 deletions ...main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,29 +172,24 @@ protected boolean executeOnce() throws Exception {
// We have to parse the privilege first, to avoid passing no-privilege data to processor
if (event instanceof PipeTsFileInsertionEvent
&& ((PipeTsFileInsertionEvent) event).shouldParse4Privilege()) {
try (final PipeTsFileInsertionEvent tsFileInsertionEvent =
(PipeTsFileInsertionEvent) event) {
final AtomicReference<Exception> ex = new AtomicReference<>();
tsFileInsertionEvent.consumeTabletInsertionEventsWithRetry(
event1 -> {
try {
pipeProcessor.process(event1, outputEventCollector);
} catch (PipeRuntimeOutOfMemoryCriticalException e) {
throw e;
} catch (Exception e) {
ex.set(e);
}
},
"PipeProcessorSubtask::executeOnce");
if (tsFileInsertionEvent.isGeneratedByHistoricalExtractor()) {
PipeTerminateEvent.markHistoricalTsFileSplit(
tsFileInsertionEvent.getPipeName(),
tsFileInsertionEvent.getCreationTime(),
regionId);
}
if (ex.get() != null) {
throw ex.get();
}
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);
}
} else {
pipeProcessor.process((TsFileInsertionEvent) event, outputEventCollector);
Expand Down
101 changes: 84 additions & 17 deletions .../src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,13 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent
private final AtomicBoolean isClosed;
private final AtomicReference<TsFileInsertionEventParser> eventParser;
private final AtomicBoolean isTsFileParserMemoryReserved = new AtomicBoolean(false);
private final AtomicReference<Iterator<TabletInsertionEvent>> tabletInsertionEventIterator =
new AtomicReference<>();
private final AtomicReference<PipeRawTabletInsertionEvent> pendingTabletInsertionEvent =
new AtomicReference<>();
private final AtomicInteger parsedTabletInsertionEventCount = new AtomicInteger(0);
private final AtomicBoolean isTsFileParsingCompleted = new AtomicBoolean(false);
private final AtomicLong parsedPointCountForCount = new AtomicLong(0);

// The point count of the TsFile. Used for metrics on IoTConsensusV2' receiver side.
// May be updated after it is flushed. Should be negative if not set.
Expand Down Expand Up @@ -747,26 +754,68 @@ public interface TabletInsertionEventConsumer {

public void consumeTabletInsertionEventsWithRetry(
final TabletInsertionEventConsumer consumer, final String callerName) throws Exception {
int tabletEventCount = 0;
try {
final Iterable<TabletInsertionEvent> iterable = toTabletInsertionEvents();
final Iterator<TabletInsertionEvent> iterator = iterable.iterator();
while (iterator.hasNext()) {
final TabletInsertionEvent parsedEvent = iterator.next();
tabletEventCount++;
while (true) {
final PipeRawTabletInsertionEvent parsedEvent =
getNextTabletInsertionEventFromSavedProgress();
if (parsedEvent == null) {
isTsFileParsingCompleted.set(true);
releaseTsFileParserMemoryIfReserved();
return;
}
consumeParsedTabletInsertionEventWithRetry(
consumer, callerName, tabletEventCount, parsedEvent);
consumer, callerName, parsedTabletInsertionEventCount.get(), parsedEvent);
pendingTabletInsertionEvent.compareAndSet(parsedEvent, null);
}
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
close();
// Yield the active parser slot to the next pipe while retaining the iterator and current
// tablet. The next retry resumes from this exact tablet instead of reparsing the TsFile.
releaseTsFileParserMemoryIfReserved();
LOGGER.warn(
DataNodePipeMessages.FAILED_TO_ALLOCATE_MEMORY_FOR_PARSING_TSFILE,
callerName,
getTsFile(),
tabletEventCount,
parsedTabletInsertionEventCount.get(),
e);
throw e;
} catch (final Exception e) {
releaseTsFileParserMemoryIfReserved();
throw e;
}
}

private PipeRawTabletInsertionEvent getNextTabletInsertionEventFromSavedProgress()
throws Exception {
if (isTsFileParsingCompleted.get()) {
return null;
}

// Reacquire parser memory after a previous failure yielded the active parser slot.
waitForResourceEnough4Parsing((long) ((1 + Math.random()) * 20 * 1000));

final PipeRawTabletInsertionEvent pendingEvent = pendingTabletInsertionEvent.get();
if (pendingEvent != null) {
return pendingEvent;
}

Iterator<TabletInsertionEvent> iterator = tabletInsertionEventIterator.get();
if (iterator == null) {
if (!waitForTsFileClose()) {
LOGGER.warn(DataNodePipeMessages.PIPE_SKIPPING_TEMPORARY_TSFILE_S_PARSING_WHICH, tsFile);
return null;
}
iterator = initEventParser().toTabletInsertionEvents().iterator();
tabletInsertionEventIterator.set(iterator);
}

if (!iterator.hasNext()) {
return null;
}

final PipeRawTabletInsertionEvent nextEvent = (PipeRawTabletInsertionEvent) iterator.next();
pendingTabletInsertionEvent.set(nextEvent);
parsedTabletInsertionEventCount.incrementAndGet();
return nextEvent;
}

private void consumeParsedTabletInsertionEventWithRetry(
Expand All @@ -788,21 +837,34 @@ private void consumeParsedTabletInsertionEventWithRetry(
}
if (memoryManager.shouldReleaseTsFileParserOnOutOfMemory(
firstOutOfMemoryTimeInMs, ++retryCount)) {
releaseParsedTabletEvent(parsedEvent);
throw e;
}
logParserRetryOnOutOfMemory(callerName, tabletEventCount, retryCount, e);
try {
Thread.sleep(PipeConfig.getInstance().getPipeMemoryAllocateRetryIntervalInMs());
Thread.sleep(getParserRetryBackoffInMs(retryCount));
} catch (final InterruptedException interruptedException) {
Thread.currentThread().interrupt();
releaseParsedTabletEvent(parsedEvent);
throw e;
}
}
}
}

private long getParserRetryBackoffInMs(final int retryCount) {
final long initialBackoffInMs =
Math.max(1, PipeConfig.getInstance().getPipeMemoryAllocateRetryIntervalInMs());
final int maxRetries = Math.max(1, PipeConfig.getInstance().getPipeMemoryAllocateMaxRetries());
final long maxBackoffInMs =
initialBackoffInMs > Long.MAX_VALUE / maxRetries
? Long.MAX_VALUE
: initialBackoffInMs * maxRetries;
final int shift = Math.min(30, Math.max(0, retryCount - 1));
if (initialBackoffInMs > (maxBackoffInMs >> shift)) {
return maxBackoffInMs;
}
return Math.min(maxBackoffInMs, initialBackoffInMs << shift);
}

private void logParserRetryOnOutOfMemory(
final String callerName,
final int tabletEventCount,
Expand Down Expand Up @@ -982,21 +1044,21 @@ private TsFileInsertionEventParser initEventParser() {
}

public long count(final boolean skipReportOnCommit) throws Exception {
AtomicLong count = new AtomicLong();

if (shouldParseTime()) {
try {
consumeTabletInsertionEventsWithRetry(
event -> {
count.addAndGet(event.count());
parsedPointCountForCount.addAndGet(event.count());
if (skipReportOnCommit) {
event.skipReportOnCommit();
}
},
"PipeTsFileInsertionEvent::count");
return count.get();
return parsedPointCountForCount.getAndSet(0);
} finally {
close();
if (isTsFileParsingCompleted.get()) {
close();
}
}
}

Expand All @@ -1009,6 +1071,11 @@ public long count(final boolean skipReportOnCommit) throws Exception {
/** Release the resource of {@link TsFileInsertionEventParser}. */
@Override
public void close() {
tabletInsertionEventIterator.set(null);
releaseParsedTabletEvent(pendingTabletInsertionEvent.getAndSet(null));
parsedTabletInsertionEventCount.set(0);
parsedPointCountForCount.set(0);
isTsFileParsingCompleted.set(false);
eventParser.getAndUpdate(
parser -> {
if (Objects.nonNull(parser)) {
Expand Down
39 changes: 18 additions & 21 deletions ...tanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -542,32 +542,29 @@ private Map<String, Pair<Long, ByteBuffer>> processRow(
public void process(
final TsFileInsertionEvent tsFileInsertionEvent, final EventCollector eventCollector)
throws Exception {
try {
if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
final AtomicReference<Exception> ex = new AtomicReference<>();
((PipeTsFileInsertionEvent) tsFileInsertionEvent)
.consumeTabletInsertionEventsWithRetry(
event -> {
try {
process(event, eventCollector);
} catch (PipeRuntimeOutOfMemoryCriticalException e) {
throw e;
} catch (Exception e) {
ex.set(e);
}
},
"AggregateProcessor::process");
if (ex.get() != null) {
throw ex.get();
}
} else {
if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
((PipeTsFileInsertionEvent) tsFileInsertionEvent)
.consumeTabletInsertionEventsWithRetry(
event -> {
try {
process(event, eventCollector);
} catch (PipeRuntimeOutOfMemoryCriticalException e) {
throw e;
} catch (Exception e) {
throw new PipeException(e.getMessage(), e);
}
},
"AggregateProcessor::process");
tsFileInsertionEvent.close();
} else {
try {
for (final TabletInsertionEvent tabletInsertionEvent :
tsFileInsertionEvent.toTabletInsertionEvents()) {
process(tabletInsertionEvent, eventCollector);
}
} finally {
tsFileInsertionEvent.close();
}
} finally {
tsFileInsertionEvent.close();
}
// The timeProgressIndex shall only be reported by the output events
// whose progressIndex is bounded with tablet events
Expand Down
40 changes: 19 additions & 21 deletions .../src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.apache.iotdb.pipe.api.exception.PipeException;

import org.apache.tsfile.common.constant.TsFileConstant;

Expand Down Expand Up @@ -149,32 +150,29 @@ protected abstract void processRow(
public void process(TsFileInsertionEvent tsFileInsertionEvent, EventCollector eventCollector)
throws Exception {
if (shouldSplitFile) {
try {
if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
final AtomicReference<Exception> ex = new AtomicReference<>();
((PipeTsFileInsertionEvent) tsFileInsertionEvent)
.consumeTabletInsertionEventsWithRetry(
event -> {
try {
process(event, eventCollector);
} catch (PipeRuntimeOutOfMemoryCriticalException e) {
throw e;
} catch (Exception e) {
ex.set(e);
}
},
"DownSamplingProcessor::process");
if (ex.get() != null) {
throw ex.get();
}
} else {
if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
((PipeTsFileInsertionEvent) tsFileInsertionEvent)
.consumeTabletInsertionEventsWithRetry(
event -> {
try {
process(event, eventCollector);
} catch (PipeRuntimeOutOfMemoryCriticalException e) {
throw e;
} catch (Exception e) {
throw new PipeException(e.getMessage(), e);
}
},
"DownSamplingProcessor::process");
tsFileInsertionEvent.close();
} else {
try {
for (final TabletInsertionEvent tabletInsertionEvent :
tsFileInsertionEvent.toTabletInsertionEvents()) {
process(tabletInsertionEvent, eventCollector);
}
} finally {
tsFileInsertionEvent.close();
}
} finally {
tsFileInsertionEvent.close();
}
} else {
eventCollector.collect(tsFileInsertionEvent);
Expand Down
21 changes: 9 additions & 12 deletions ...atanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,18 +145,15 @@ public void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception
return;
}

try {
((PipeTsFileInsertionEvent) tsFileInsertionEvent)
.consumeTabletInsertionEventsWithRetry(
event -> {
// Skip report if any tablet events is added
((PipeTsFileInsertionEvent) tsFileInsertionEvent).skipReportOnCommit();
transfer(event);
},
"WebSocketConnector::transfer");
} finally {
tsFileInsertionEvent.close();
}
((PipeTsFileInsertionEvent) tsFileInsertionEvent)
.consumeTabletInsertionEventsWithRetry(
event -> {
// Skip report if any tablet events is added
((PipeTsFileInsertionEvent) tsFileInsertionEvent).skipReportOnCommit();
transfer(event);
},
"WebSocketConnector::transfer");
tsFileInsertionEvent.close();
}

@Override
Expand Down
Loading
Loading