Skip to content

IoTV2: Remove events that do not retry in leader's transfer buffer & fix concurrency bug of retryQueue & fix retry interval & refine log#15684

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

Merged
jt2594838 merged 2 commits into from
Jun 16, 2025
Merged

IoTV2: Remove events that do not retry in leader's transfer buffer & fix concurrency bug of retryQueue & fix retry interval & refine log #15684

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
002860b
remove events that do not retry
Pengzna Jun 9, 2025
8f7e50f
refine log & fix retry interval & fix concurrency issue of retryQueue
Pengzna Jun 12, 2025
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
28 changes: 19 additions & 9 deletions ...rg/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -72,10 +72,10 @@
import java.util.Comparator;
import java.util.Iterator;
import java.util.Objects;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
Expand All@@ -101,7 +101,7 @@ public class PipeConsensusAsyncConnector extends IoTDBConnector implements Conse
private static final long PIPE_CONSENSUS_EVENT_ENQUEUE_TIMEOUT_IN_MS =
IOTDB_CONFIG.getConnectionTimeoutInMS() / 6;
private final Queue<EnrichedEvent> retryEventQueue =
new PriorityQueue<>(
new PriorityBlockingQueue<>(
IOTDB_CONFIG.getIotConsensusV2PipelineSize(),
Comparator.comparingLong(EnrichedEvent::getReplicateIndexForIoTV2));
// We use enrichedEvent here to make use of EnrichedEvent.equalsInPipeConsensus
Expand DownExpand Up@@ -535,9 +535,10 @@ private void asyncTransferQueuedEventsIfNecessary() {
? peekedEvent.getRetryInterval()
: 0L;
LOGGER.info(
"PipeConsensus-ConsensusGroup-{}: retry with interval {} for {}",
"PipeConsensus-ConsensusGroup-{}: retry with interval {} for index {} {}",
consensusGroupId,
retryInterval,
peekedEvent.getReplicateIndexForIoTV2(),
peekedEvent);
Comment thread
jt2594838 marked this conversation as resolved.
// need to retry in background service, otherwise the retryInterval will block the sender
// procedure.
Expand DownExpand Up@@ -616,7 +617,7 @@ private void retryTransfer(final PipeDeleteDataNodeEvent deleteDataNodeEvent) {
* @param event event to retry
*/
@SuppressWarnings("java:S899")
public void addFailureEventToRetryQueue(final EnrichedEvent event) {
public synchronized void addFailureEventToRetryQueue(final EnrichedEvent event) {
Comment thread
jt2594838 marked this conversation as resolved.
if (event.isReleased()) {
return;
}
Expand All@@ -630,11 +631,20 @@ public void addFailureEventToRetryQueue(final EnrichedEvent event) {
return;
}

retryEventQueue.offer(event);
LOGGER.info(
"PipeConsensus-ConsensusGroup-{}: Event {} transfer failed, will be added to retry queue.",
consensusGroupId,
event);
boolean res = retryEventQueue.offer(event);
if (res) {
LOGGER.info(
"PipeConsensus-ConsensusGroup-{}: Event {} replicate index {} transfer failed, will be added to retry queue.",
consensusGroupId,
event,
event.getReplicateIndexForIoTV2());
} else {
LOGGER.warn(
"PipeConsensus-ConsensusGroup-{}: Event {} replicate index {} transfer failed, added to retry queue failed, this event will be ignored.",
consensusGroupId,
event,
event.getReplicateIndexForIoTV2());
}
Comment thread
jt2594838 marked this conversation as resolved.

if (isClosed.get()) {
event.clearReferenceCount(PipeConsensusAsyncConnector.class.getName());
Expand Down
4 changes: 3 additions & 1 deletion ...tdb/db/pipe/connector/protocol/pipeconsensus/handler/PipeConsensusDeleteEventHandler.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -87,8 +87,10 @@ public void onComplete(TPipeConsensusTransferResp response) {
LOGGER.info(
"DeleteNodeTransfer: no.{} event successfully processed!",
event.getReplicateIndexForIoTV2());
connector.removeEventFromBuffer(event);
}
// if code flow reach here, meaning the file will not be resent and will be ignored.
// events that don't need to be retried will be removed from the buffer
connector.removeEventFromBuffer(event);

long duration = System.nanoTime() - createTime;
metric.recordConnectorWalTransferTimer(duration);
Expand Down
5 changes: 4 additions & 1 deletion ...pe/connector/protocol/pipeconsensus/handler/PipeConsensusTabletInsertionEventHandler.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -95,9 +95,12 @@ public void onComplete(TPipeConsensusTransferResp response) {
LOGGER.info(
"InsertNodeTransfer: no.{} event successfully processed!",
((EnrichedEvent) event).getReplicateIndexForIoTV2());
connector.removeEventFromBuffer((EnrichedEvent) event);
}

// if code flow reach here, meaning the file will not be resent and will be ignored.
// events that don't need to be retried will be removed from the buffer
connector.removeEventFromBuffer((EnrichedEvent) event);

long duration = System.nanoTime() - createTime;
metric.recordConnectorWalTransferTimer(duration);
} catch (Exception e) {
Expand Down
10 changes: 5 additions & 5 deletions ...pe/connector/protocol/pipeconsensus/handler/PipeConsensusTsFileInsertionEventHandler.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -210,9 +210,9 @@ public void onComplete(final TPipeConsensusTransferResp response) {
tsFile.getName());
}

if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
connector.removeEventFromBuffer(event);
}
// if code flow reach here, meaning the file will not be resent and will be ignored.
// events that don't need to be retried will be removed from the buffer
connector.removeEventFromBuffer(event);
} catch (final Exception e) {
onError(e);
return;
Expand DownExpand Up@@ -295,10 +295,10 @@ public void onError(final Exception exception) {

if (RetryUtils.needRetryWithIncreasingInterval(exception)) {
// just in case for overflow
if (event.getRetryInterval() << 2 <= 0) {
if (event.getRetryInterval() << 1 <= 0) {
event.setRetryInterval(1000L * 20);
} else {
event.setRetryInterval(Math.min(1000L * 20, event.getRetryInterval() << 2));
event.setRetryInterval(Math.min(1000L * 20, event.getRetryInterval() << 1));
}
}

Expand Down
6 changes: 3 additions & 3 deletions ...anode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -61,7 +61,7 @@ public static PipeRealtimeEvent createRealtimeEvent(
&& PipeConsensusProcessor.isShouldReplicate(tsFileInsertionEvent)) {
tsFileInsertionEvent.setReplicateIndexForIoTV2(
ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(dataRegionId));
LOGGER.debug(
LOGGER.info(
"[Region{}]Set {} for event {}",
dataRegionId,
tsFileInsertionEvent.getReplicateIndexForIoTV2(),
Expand DownExpand Up@@ -103,7 +103,7 @@ public static PipeRealtimeEvent createRealtimeEvent(
&& PipeConsensusProcessor.isShouldReplicate(insertionEvent)) {
insertionEvent.setReplicateIndexForIoTV2(
ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(dataRegionId));
LOGGER.debug(
LOGGER.info(
"[Region{}]Set {} for event {}",
dataRegionId,
insertionEvent.getReplicateIndexForIoTV2(),
Expand All@@ -130,7 +130,7 @@ public static PipeRealtimeEvent createRealtimeEvent(
&& PipeConsensusProcessor.isShouldReplicate(deleteDataNodeEvent)) {
deleteDataNodeEvent.setReplicateIndexForIoTV2(
ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(dataRegionId));
LOGGER.debug(
LOGGER.info(
"[Region{}]Set {} for event {}",
dataRegionId,
deleteDataNodeEvent.getReplicateIndexForIoTV2(),
Expand Down
2 changes: 1 addition & 1 deletion ...e/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -879,7 +879,7 @@ private Event supplyTsFileEvent(final TsFileResource resource) {
event.setReplicateIndexForIoTV2(
ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(
resource.getDataRegionId()));
LOGGER.debug(
LOGGER.info(
"[Region{}]Set {} for event {}",
resource.getDataRegionId(),
event.getReplicateIndexForIoTV2(),
Expand Down
4 changes: 2 additions & 2 deletions iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -435,8 +435,8 @@ public long getRetryInterval() {
return this.retryInterval;
}

public long setRetryInterval(final long retryInterval) {
return retryInterval;
public void setRetryInterval(final long retryInterval) {
this.retryInterval = retryInterval;
}

public CommitterKey getCommitterKey() {
Expand Down