Skip to content

Commit ed52dbd

Browse files
DanielWang2035SyncGithubBot
DanielWang2035
authored and
SyncGithubBot
committed
[To dev/1.3] Pipe: Add retry when TsFile parsing failed to avoid race among processor threads (apache#15624, apache#15644) (apache#15659)
* Pipe: Add retry when TsFile parsing failed to avoid race among processor threads (apache#15624) (cherry picked from commit 7ad757f) * Pipe: Add retry when TsFile parsing failed to avoid race among processor threads (apache#15624, apache#15644) * refactor * refactor * refactor (cherry picked from commit c28e50f)
1 parent 8a55f59 commit ed52dbd

6 files changed

Lines changed: 123 additions & 40 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,6 @@
3434
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
3535
import org.apache.iotdb.pipe.api.collector.EventCollector;
3636
import org.apache.iotdb.pipe.api.event.Event;
37-
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
3837
import org.apache.iotdb.pipe.api.exception.PipeException;
3938

4039
import org.slf4j.Logger;
@@ -141,9 +140,8 @@ private void parseAndCollectEvent(final PipeTsFileInsertionEvent sourceEvent) th
141140
}
142141

143142
try {
144-
for (final TabletInsertionEvent parsedEvent : sourceEvent.toTabletInsertionEvents()) {
145-
collectParsedRawTableEvent((PipeRawTabletInsertionEvent) parsedEvent);
146-
}
143+
sourceEvent.consumeTabletInsertionEventsWithRetry(
144+
this::collectParsedRawTableEvent, "PipeEventCollector::parseAndCollectEvent");
147145
} finally {
148146
sourceEvent.close();
149147
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/websocket/WebSocketConnector.java‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -139,11 +139,14 @@ public void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception
139139
}
140140

141141
try {
142-
for (TabletInsertionEvent event : tsFileInsertionEvent.toTabletInsertionEvents()) {
143-
// Skip report if any tablet events is added
144-
((PipeTsFileInsertionEvent) tsFileInsertionEvent).skipReportOnCommit();
145-
transfer(event);
146-
}
142+
((PipeTsFileInsertionEvent) tsFileInsertionEvent)
143+
.consumeTabletInsertionEventsWithRetry(
144+
event -> {
145+
// Skip report if any tablet events is added
146+
((PipeTsFileInsertionEvent) tsFileInsertionEvent).skipReportOnCommit();
147+
transfer(event);
148+
},
149+
"WebSocketConnector::transfer");
147150
} finally {
148151
tsFileInsertionEvent.close();
149152
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java‎

Lines changed: 56 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
2323
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
24+
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
2425
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
2526
import org.apache.iotdb.commons.pipe.config.PipeConfig;
2627
import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
@@ -50,11 +51,13 @@
5051
import java.io.File;
5152
import java.io.IOException;
5253
import java.util.Collections;
54+
import java.util.Iterator;
5355
import java.util.Map;
5456
import java.util.Objects;
5557
import java.util.Set;
5658
import java.util.concurrent.atomic.AtomicBoolean;
5759
import java.util.concurrent.atomic.AtomicInteger;
60+
import java.util.concurrent.atomic.AtomicLong;
5861
import java.util.concurrent.atomic.AtomicReference;
5962

6063
public class PipeTsFileInsertionEvent extends EnrichedEvent
@@ -413,6 +416,49 @@ public boolean mayEventPathsOverlappedWithPattern() {
413416

414417
/////////////////////////// TsFileInsertionEvent ///////////////////////////
415418

419+
@FunctionalInterface
420+
public interface TabletInsertionEventConsumer {
421+
void consume(final PipeRawTabletInsertionEvent event);
422+
}
423+
424+
public void consumeTabletInsertionEventsWithRetry(
425+
final TabletInsertionEventConsumer consumer, final String callerName) throws PipeException {
426+
final Iterable<TabletInsertionEvent> iterable = toTabletInsertionEvents();
427+
final Iterator<TabletInsertionEvent> iterator = iterable.iterator();
428+
int tabletEventCount = 0;
429+
while (iterator.hasNext()) {
430+
final TabletInsertionEvent parsedEvent = iterator.next();
431+
tabletEventCount++;
432+
int retryCount = 0;
433+
while (true) {
434+
// If failed due do insufficient memory, retry until success to avoid race among multiple
435+
// processor threads
436+
try {
437+
consumer.consume((PipeRawTabletInsertionEvent) parsedEvent);
438+
break;
439+
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
440+
if (retryCount++ % 100 == 0) {
441+
LOGGER.warn(
442+
"{}: failed to allocate memory for parsing TsFile {}, tablet event no. {}, retry count is {}, will keep retrying.",
443+
callerName,
444+
getTsFile(),
445+
tabletEventCount,
446+
retryCount,
447+
e);
448+
} else if (LOGGER.isDebugEnabled()) {
449+
LOGGER.debug(
450+
"{}: failed to allocate memory for parsing TsFile {}, tablet event no. {}, retry count is {}, will keep retrying.",
451+
callerName,
452+
getTsFile(),
453+
tabletEventCount,
454+
retryCount,
455+
e);
456+
}
457+
}
458+
}
459+
}
460+
}
461+
416462
@Override
417463
public Iterable<TabletInsertionEvent> toTabletInsertionEvents() throws PipeException {
418464
// 20 - 40 seconds for waiting
@@ -528,18 +574,19 @@ private TsFileInsertionDataContainer initDataContainer() {
528574
}
529575

530576
public long count(final boolean skipReportOnCommit) throws IOException {
531-
long count = 0;
577+
AtomicLong count = new AtomicLong();
532578

533579
if (shouldParseTime()) {
534580
try {
535-
for (final TabletInsertionEvent event : toTabletInsertionEvents()) {
536-
final PipeRawTabletInsertionEvent rawEvent = ((PipeRawTabletInsertionEvent) event);
537-
count += rawEvent.count();
538-
if (skipReportOnCommit) {
539-
rawEvent.skipReportOnCommit();
540-
}
541-
}
542-
return count;
581+
consumeTabletInsertionEventsWithRetry(
582+
event -> {
583+
count.addAndGet(event.count());
584+
if (skipReportOnCommit) {
585+
event.skipReportOnCommit();
586+
}
587+
},
588+
"PipeTsFileInsertionEvent::count");
589+
return count.get();
543590
} finally {
544591
close();
545592
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java‎

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -512,9 +512,26 @@ public void process(
512512
final TsFileInsertionEvent tsFileInsertionEvent, final EventCollector eventCollector)
513513
throws Exception {
514514
try {
515-
for (final TabletInsertionEvent tabletInsertionEvent :
516-
tsFileInsertionEvent.toTabletInsertionEvents()) {
517-
process(tabletInsertionEvent, eventCollector);
515+
if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
516+
final AtomicReference<Exception> ex = new AtomicReference<>();
517+
((PipeTsFileInsertionEvent) tsFileInsertionEvent)
518+
.consumeTabletInsertionEventsWithRetry(
519+
event -> {
520+
try {
521+
process(event, eventCollector);
522+
} catch (Exception e) {
523+
ex.set(e);
524+
}
525+
},
526+
"AggregateProcessor::process");
527+
if (ex.get() != null) {
528+
throw ex.get();
529+
}
530+
} else {
531+
for (final TabletInsertionEvent tabletInsertionEvent :
532+
tsFileInsertionEvent.toTabletInsertionEvents()) {
533+
process(tabletInsertionEvent, eventCollector);
534+
}
518535
}
519536
} finally {
520537
tsFileInsertionEvent.close();

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingProcessor.java‎

Lines changed: 21 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskProcessorRuntimeEnvironment;
2424
import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
2525
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
26+
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
2627
import org.apache.iotdb.db.storageengine.StorageEngine;
2728
import org.apache.iotdb.pipe.api.PipeProcessor;
2829
import org.apache.iotdb.pipe.api.access.Row;
@@ -45,7 +46,6 @@
4546
import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_DOWN_SAMPLING_SPLIT_FILE_KEY;
4647

4748
public abstract class DownSamplingProcessor implements PipeProcessor {
48-
4949
protected long memoryLimitInBytes;
5050

5151
protected boolean shouldSplitFile;
@@ -149,9 +149,26 @@ public void process(TsFileInsertionEvent tsFileInsertionEvent, EventCollector ev
149149
throws Exception {
150150
if (shouldSplitFile) {
151151
try {
152-
for (final TabletInsertionEvent tabletInsertionEvent :
153-
tsFileInsertionEvent.toTabletInsertionEvents()) {
154-
process(tabletInsertionEvent, eventCollector);
152+
if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
153+
final AtomicReference<Exception> ex = new AtomicReference<>();
154+
((PipeTsFileInsertionEvent) tsFileInsertionEvent)
155+
.consumeTabletInsertionEventsWithRetry(
156+
event -> {
157+
try {
158+
process(event, eventCollector);
159+
} catch (Exception e) {
160+
ex.set(e);
161+
}
162+
},
163+
"DownSamplingProcessor::process");
164+
if (ex.get() != null) {
165+
throw ex.get();
166+
}
167+
} else {
168+
for (final TabletInsertionEvent tabletInsertionEvent :
169+
tsFileInsertionEvent.toTabletInsertionEvents()) {
170+
process(tabletInsertionEvent, eventCollector);
171+
}
155172
}
156173
} finally {
157174
tsFileInsertionEvent.close();

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java‎

Lines changed: 16 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121

2222
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
2323
import org.apache.iotdb.db.pipe.connector.payload.evolvable.batch.PipeTabletEventTsFileBatch;
24-
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
2524
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
2625
import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingTsFileQueue;
2726
import org.apache.iotdb.db.subscription.event.SubscriptionEvent;
@@ -95,20 +94,22 @@ protected void onTabletInsertionEvent(final TabletInsertionEvent event) {
9594
protected void onTsFileInsertionEvent(final TsFileInsertionEvent event) {
9695
// TODO: parse tsfile event on the fly like SubscriptionPipeTabletEventBatch
9796
try {
98-
for (final TabletInsertionEvent parsedEvent : event.toTabletInsertionEvents()) {
99-
if (!((PipeRawTabletInsertionEvent) parsedEvent)
100-
.increaseReferenceCount(this.getClass().getName())) {
101-
LOGGER.warn(
102-
"SubscriptionPipeTsFileEventBatch: Failed to increase the reference count of event {}, skipping it.",
103-
((PipeRawTabletInsertionEvent) parsedEvent).coreReportMessage());
104-
} else {
105-
try {
106-
batch.onEvent(parsedEvent);
107-
} catch (final Exception ignored) {
108-
// no exceptions will be thrown
109-
}
110-
}
111-
}
97+
((PipeTsFileInsertionEvent) event)
98+
.consumeTabletInsertionEventsWithRetry(
99+
event1 -> {
100+
if (!event1.increaseReferenceCount(this.getClass().getName())) {
101+
LOGGER.warn(
102+
"SubscriptionPipeTsFileEventBatch: Failed to increase the reference count of event {}, skipping it.",
103+
event1.coreReportMessage());
104+
} else {
105+
try {
106+
batch.onEvent(event1);
107+
} catch (final Exception ignored) {
108+
// no exceptions will be thrown
109+
}
110+
}
111+
},
112+
"SubscriptionPipeTsFileEventBatch::onTsFileInsertionEvent");
112113
} finally {
113114
try {
114115
event.close();

0 commit comments

Comments
 (0)