jt2594838 commented on code in PR #18261:
URL: https://github.com/apache/iotdb/pull/18261#discussion_r3627039142
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java:
##########
@@ -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);
Review Comment:
What does `Math.min(30` mean
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java:
##########
@@ -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));
Review Comment:
Backoff does not work on this?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java:
##########
@@ -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);
}
Review Comment:
Double-check the risk of resource leak after this modification.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
