You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/04/27 10:50:17 UTC
(iotdb) branch drop-pipe-fix updated (cf718f6e444 -> d8114969d0a)
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a change to branch drop-pipe-fix
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from cf718f6e444 Use MemoizedCheckedSupplier for COPY TO folder manager & keep compatibility with another branch (#17559)
new 4ec94cd7bf7 drop-1
new 543efacc830 wd
new a3b7ea0bcd4 drop
new d8114969d0a fix
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../agent/task/connection/PipeEventCollector.java | 18 ++-
.../sink/PipeRealtimePriorityBlockingQueue.java | 13 +-
.../agent/task/subtask/sink/PipeSinkSubtask.java | 13 +-
.../subtask/sink/PipeSinkSubtaskLifeCycle.java | 5 +-
.../task/subtask/sink/PipeSinkSubtaskManager.java | 2 +-
.../evolvable/batch/PipeTabletEventBatch.java | 7 +-
.../batch/PipeTransferBatchReqBuilder.java | 9 +-
.../thrift/async/IoTDBDataRegionAsyncSink.java | 51 ++++++--
.../thrift/sync/IoTDBDataRegionSyncSink.java | 5 +-
.../websocket/WebSocketConnectorServer.java | 144 +++++++++++++++++----
.../sink/protocol/websocket/WebSocketSink.java | 11 +-
.../subtask/SubscriptionSinkSubtaskLifeCycle.java | 3 +-
.../subtask/SubscriptionSinkSubtaskManager.java | 2 +-
.../task/connection/PipeEventCollectorTest.java | 97 ++++++++++++++
.../sink/PipeSinkSubtaskTest.java} | 39 ++++--
.../apache/iotdb/db/pipe/sink/PipeSinkTest.java | 105 +++++++++++++++
.../opc_security/8443_12686/iotdb-server.pfx | Bin 0 -> 2942 bytes
.../task/connection/BlockingPendingQueue.java | 56 +++++++-
.../commons/pipe/sink/protocol/IoTDBSink.java | 5 +-
.../protocol/PipeConnectorWithEventDiscard.java | 6 +-
20 files changed, 504 insertions(+), 87 deletions(-)
create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java
copy iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/{PipeSinkSubtaskExecutorTest.java => subtask/sink/PipeSinkSubtaskTest.java} (52%)
create mode 100644 iotdb-core/datanode/src/test/resources/opc_security/8443_12686/iotdb-server.pfx
copy iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionMessageHandler.java => iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/PipeConnectorWithEventDiscard.java (81%)
(iotdb) 02/04: wd
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch drop-pipe-fix
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 543efacc830755c006836575e65fb2e1ab7ed09d
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Mon Apr 27 12:22:13 2026 +0800
wd
---
.../agent/task/subtask/sink/PipeSinkSubtask.java | 5 +-
.../websocket/WebSocketConnectorServer.java | 143 +++++++++++++++++----
.../sink/protocol/websocket/WebSocketSink.java | 11 +-
.../task/subtask/sink/PipeSinkSubtaskTest.java | 61 +++++++++
.../apache/iotdb/db/pipe/sink/PipeSinkTest.java | 42 ++++++
.../commons/pipe/sink/protocol/IoTDBSink.java | 2 +-
.../protocol/PipeConnectorWithEventDiscard.java | 25 ++++
7 files changed, 259 insertions(+), 30 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
index f6008822e61..dd31d5b5fd8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSink;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.commons.pipe.sink.protocol.PipeConnectorWithEventDiscard;
import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.event.UserDefinedEnrichedEvent;
@@ -248,8 +249,8 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
decreaseHighPriorityTaskCount();
}
- if (outputPipeSink instanceof IoTDBSink) {
- ((IoTDBSink) outputPipeSink)
+ if (outputPipeSink instanceof PipeConnectorWithEventDiscard) {
+ ((PipeConnectorWithEventDiscard) outputPipeSink)
.discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketConnectorServer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketConnectorServer.java
index 491d40a1e45..6b402b27026 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketConnectorServer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketConnectorServer.java
@@ -39,6 +39,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -56,6 +57,7 @@ public class WebSocketConnectorServer extends WebSocketServer {
// Map<pipeName, Map<eventId, Tuple<connector, event>>>
private final ConcurrentHashMap<String, ConcurrentHashMap<Long, EventWaitingForAck>>
eventsWaitingForAck = new ConcurrentHashMap<>();
+ private final Set<String> droppedPipeTaskKeys = ConcurrentHashMap.newKeySet();
private final BidiMap<String, WebSocket> router =
new DualTreeBidiMap<String, WebSocket>(null, Comparator.comparing(Object::hashCode)) {};
@@ -97,13 +99,8 @@ public class WebSocketConnectorServer extends WebSocketServer {
eventWrappers = new ArrayList<>(eventTransferQueue);
eventTransferQueue.clear();
}
- eventWrappers.forEach(
- (eventWrapper) -> {
- if (eventWrapper.event instanceof EnrichedEvent) {
- ((EnrichedEvent) eventWrapper.event)
- .decreaseReferenceCount(WebSocketConnectorServer.class.getName(), false);
- }
- });
+ eventWrappers.forEach(eventWrapper -> discardEvent(eventWrapper.event));
+ eventWrappers.clear();
synchronized (eventTransferQueue) {
eventTransferQueue.notifyAll();
}
@@ -113,13 +110,35 @@ public class WebSocketConnectorServer extends WebSocketServer {
if (eventsWaitingForAck.containsKey(pipeName)) {
eventsWaitingForAck
.remove(pipeName)
- .forEach(
- (eventId, eventWrapper) -> {
- if (eventWrapper.event instanceof EnrichedEvent) {
- ((EnrichedEvent) eventWrapper.event)
- .decreaseReferenceCount(WebSocketConnectorServer.class.getName(), false);
- }
- });
+ .forEach((eventId, eventWrapper) -> discardEvent(eventWrapper.event));
+ }
+
+ droppedPipeTaskKeys.removeIf(key -> key.startsWith(pipeName + "_"));
+ }
+
+ public synchronized void discardEventsOfPipe(
+ final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) {
+ droppedPipeTaskKeys.add(generatePipeTaskKey(pipeNameToDrop, creationTimeToDrop, regionId));
+
+ final PriorityBlockingQueue<EventWaitingForTransfer> eventTransferQueue =
+ eventsWaitingForTransfer.get(pipeNameToDrop);
+ if (eventTransferQueue != null) {
+ eventTransferQueue.removeIf(
+ eventWrapper ->
+ discardIfMatches(
+ eventWrapper.event, pipeNameToDrop, creationTimeToDrop, regionId));
+ synchronized (eventTransferQueue) {
+ eventTransferQueue.notifyAll();
+ }
+ }
+
+ final ConcurrentHashMap<Long, EventWaitingForAck> eventId2EventMap =
+ eventsWaitingForAck.get(pipeNameToDrop);
+ if (eventId2EventMap != null) {
+ eventId2EventMap.entrySet().removeIf(
+ entry ->
+ discardIfMatches(
+ entry.getValue().event, pipeNameToDrop, creationTimeToDrop, regionId));
}
}
@@ -300,21 +319,24 @@ public class WebSocketConnectorServer extends WebSocketServer {
}
public void addEvent(Event event, WebSocketSink connector) {
+ if (isDroppedPipe(event)) {
+ discardEvent(event);
+ return;
+ }
+
+ final String pipeName = connector.getPipeName();
final PriorityBlockingQueue<EventWaitingForTransfer> queue =
- eventsWaitingForTransfer.get(connector.getPipeName());
+ eventsWaitingForTransfer.get(pipeName);
if (queue == null) {
LOGGER.warn("The pipe {} was dropped so the event {} will be dropped.", connector, event);
- if (event instanceof EnrichedEvent) {
- ((EnrichedEvent) event)
- .decreaseReferenceCount(WebSocketConnectorServer.class.getName(), false);
- }
+ discardEvent(event);
return;
}
if (queue.size() >= 5) {
synchronized (queue) {
- while (queue.size() >= 5) {
+ while (queue.size() >= 5 && isQueueAvailable(pipeName, queue) && !isDroppedPipe(event)) {
try {
queue.wait();
} catch (InterruptedException e) {
@@ -323,15 +345,27 @@ public class WebSocketConnectorServer extends WebSocketServer {
}
}
+ if (!isQueueAvailable(pipeName, queue) || isDroppedPipe(event)) {
+ discardEvent(event);
+ return;
+ }
+
queue.put(
new EventWaitingForTransfer(eventIdGenerator.incrementAndGet(), connector, event));
return;
}
}
+ if (!isQueueAvailable(pipeName, queue) || isDroppedPipe(event)) {
+ discardEvent(event);
+ return;
+ }
+
synchronized (queue) {
queue.put(new EventWaitingForTransfer(eventIdGenerator.incrementAndGet(), connector, event));
}
+
+ queue.put(new EventWaitingForTransfer(eventIdGenerator.incrementAndGet(), connector, event));
}
private class TransferThread extends Thread {
@@ -377,6 +411,11 @@ public class WebSocketConnectorServer extends WebSocketServer {
final WebSocketSink connector = element.connector;
try {
+ if (isDroppedPipe(event)) {
+ discardEvent(event);
+ return;
+ }
+
ByteBuffer tabletBuffer;
if (event instanceof PipeRawTabletInsertionEvent) {
tabletBuffer = ((PipeRawTabletInsertionEvent) event).convertToTablet().serialize();
@@ -387,7 +426,11 @@ public class WebSocketConnectorServer extends WebSocketServer {
}
if (tabletBuffer == null) {
- connector.commit((EnrichedEvent) event);
+ if (isDroppedPipe(event)) {
+ discardEvent(event);
+ } else {
+ connector.commit((EnrichedEvent) event);
+ }
return;
}
@@ -398,11 +441,17 @@ public class WebSocketConnectorServer extends WebSocketServer {
server.broadcast(payload, Collections.singletonList(router.get(pipeName)));
+ if (isDroppedPipe(event)) {
+ discardEvent(event);
+ return;
+ }
+
final ConcurrentHashMap<Long, EventWaitingForAck> eventId2EventMap =
eventsWaitingForAck.get(pipeName);
if (eventId2EventMap == null) {
LOGGER.warn(
"The pipe {} was dropped so the event ack {} will be ignored.", pipeName, eventId);
+ discardEvent(event);
return;
}
eventId2EventMap.put(eventId, new EventWaitingForAck(connector, event));
@@ -410,13 +459,10 @@ public class WebSocketConnectorServer extends WebSocketServer {
synchronized (server) {
final PriorityBlockingQueue<EventWaitingForTransfer> queue =
eventsWaitingForTransfer.get(pipeName);
- if (queue == null) {
+ if (queue == null || isDroppedPipe(event)) {
LOGGER.warn(
"The pipe {} was dropped so the event {} will be dropped.", pipeName, eventId);
- if (event instanceof EnrichedEvent) {
- ((EnrichedEvent) event)
- .decreaseReferenceCount(WebSocketConnectorServer.class.getName(), false);
- }
+ discardEvent(event);
return;
}
@@ -465,4 +511,49 @@ public class WebSocketConnectorServer extends WebSocketServer {
this.event = event;
}
}
+
+ private boolean discardIfMatches(
+ final Event event,
+ final String pipeNameToDrop,
+ final long creationTimeToDrop,
+ final int regionId) {
+ if (!(event instanceof EnrichedEvent)) {
+ return false;
+ }
+
+ final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
+ if (!pipeNameToDrop.equals(enrichedEvent.getPipeName())
+ || creationTimeToDrop != enrichedEvent.getCreationTime()
+ || regionId != enrichedEvent.getRegionId()) {
+ return false;
+ }
+
+ discardEvent(enrichedEvent);
+ return true;
+ }
+
+ private boolean isDroppedPipe(final Event event) {
+ return event instanceof EnrichedEvent
+ && droppedPipeTaskKeys.contains(
+ generatePipeTaskKey(
+ ((EnrichedEvent) event).getPipeName(),
+ ((EnrichedEvent) event).getCreationTime(),
+ ((EnrichedEvent) event).getRegionId()));
+ }
+
+ private boolean isQueueAvailable(
+ final String pipeName, final PriorityBlockingQueue<EventWaitingForTransfer> queue) {
+ return eventsWaitingForTransfer.get(pipeName) == queue;
+ }
+
+ private static String generatePipeTaskKey(
+ final String pipeName, final long creationTime, final int regionId) {
+ return pipeName + "_" + creationTime + "_" + regionId;
+ }
+
+ private void discardEvent(final Event event) {
+ if (event instanceof EnrichedEvent) {
+ ((EnrichedEvent) event).clearReferenceCount(WebSocketSink.class.getName());
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java
index 4e593e673fd..dadeee8053d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.pipe.sink.protocol.websocket;
+import org.apache.iotdb.commons.pipe.sink.protocol.PipeConnectorWithEventDiscard;
import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
@@ -41,7 +42,7 @@ import java.util.Arrays;
import java.util.Optional;
@TreeModel
-public class WebSocketSink implements PipeConnector {
+public class WebSocketSink implements PipeConnector, PipeConnectorWithEventDiscard {
private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketSink.class);
@@ -166,6 +167,14 @@ public class WebSocketSink implements PipeConnector {
}
}
+ @Override
+ public void discardEventsOfPipe(
+ final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) {
+ if (server != null) {
+ server.discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId);
+ }
+ }
+
public void commit(EnrichedEvent enrichedEvent) {
Optional.ofNullable(enrichedEvent)
.ifPresent(event -> event.decreaseReferenceCount(WebSocketSink.class.getName(), true));
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
new file mode 100644
index 00000000000..ddfc699721b
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.agent.task.subtask.sink;
+
+import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
+import org.apache.iotdb.commons.pipe.sink.protocol.PipeConnectorWithEventDiscard;
+import org.apache.iotdb.pipe.api.PipeConnector;
+
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.withSettings;
+
+public class PipeSinkSubtaskTest {
+
+ @Test
+ public void testDiscardEventsOfPipeDelegatesToConnector() {
+ final PipeConnector connector =
+ mock(
+ PipeConnector.class,
+ withSettings().extraInterfaces(PipeConnectorWithEventDiscard.class));
+ final UnboundedBlockingPendingQueue<?> pendingQueue = mock(UnboundedBlockingPendingQueue.class);
+
+ final PipeSinkSubtask subtask =
+ Mockito.spy(
+ new PipeSinkSubtask(
+ "PipeSinkSubtaskTest",
+ System.currentTimeMillis(),
+ "data_test",
+ 0,
+ (UnboundedBlockingPendingQueue) pendingQueue,
+ connector));
+
+ try {
+ subtask.discardEventsOfPipe("pipe", 1L, 1);
+
+ verify((PipeConnectorWithEventDiscard) connector).discardEventsOfPipe("pipe", 1L, 1);
+ } finally {
+ subtask.close();
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
index 7dfd0446038..3ad262130ff 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
@@ -29,6 +29,8 @@ import org.apache.iotdb.db.pipe.sink.protocol.legacy.IoTDBLegacyPipeSink;
import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink;
import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
import org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink;
+import org.apache.iotdb.db.pipe.sink.protocol.websocket.WebSocketConnectorServer;
+import org.apache.iotdb.db.pipe.sink.protocol.websocket.WebSocketSink;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
@@ -39,6 +41,7 @@ import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.Assert;
import org.junit.Test;
+import org.mockito.Mockito;
import java.security.SecureRandom;
import java.util.Arrays;
@@ -146,6 +149,45 @@ public class PipeSinkTest {
}
}
+ @Test
+ public void testWebSocketSinkDropDoesNotRequeueDroppedPipeEvents() {
+ final String pipeName = "pipe_" + System.nanoTime();
+ final WebSocketConnectorServer server = WebSocketConnectorServer.getOrCreateInstance(0);
+ final WebSocketSink connector = Mockito.mock(WebSocketSink.class);
+ Mockito.when(connector.getPipeName()).thenReturn(pipeName);
+
+ server.register(connector);
+ try {
+ final PipeRawTabletInsertionEvent droppedEvent =
+ createPipeRawTabletInsertionEvent(pipeName, 1L, 1);
+ droppedEvent.increaseReferenceCount(WebSocketSink.class.getName());
+ droppedEvent.setCommitterKeyAndCommitId(new CommitterKey(pipeName, 1L, 1, -1), 1L);
+ server.addEvent(droppedEvent, connector);
+
+ server.discardEventsOfPipe(pipeName, 1L, 1);
+ Assert.assertTrue(droppedEvent.isReleased());
+
+ final PipeRawTabletInsertionEvent recreatedDroppedPipeEvent =
+ createPipeRawTabletInsertionEvent(pipeName, 1L, 1);
+ recreatedDroppedPipeEvent.increaseReferenceCount(WebSocketSink.class.getName());
+ recreatedDroppedPipeEvent.setCommitterKeyAndCommitId(
+ new CommitterKey(pipeName, 1L, 1, -1), 2L);
+ server.addEvent(recreatedDroppedPipeEvent, connector);
+
+ Assert.assertTrue(recreatedDroppedPipeEvent.isReleased());
+
+ final PipeRawTabletInsertionEvent recreatedPipeEvent =
+ createPipeRawTabletInsertionEvent(pipeName, 2L, 1);
+ recreatedPipeEvent.increaseReferenceCount(WebSocketSink.class.getName());
+ recreatedPipeEvent.setCommitterKeyAndCommitId(new CommitterKey(pipeName, 2L, 1, -1), 3L);
+ server.addEvent(recreatedPipeEvent, connector);
+
+ Assert.assertFalse(recreatedPipeEvent.isReleased());
+ } finally {
+ server.unregister(connector);
+ }
+ }
+
@Test
public void testOpcUaSink() {
final List<IMeasurementSchema> schemaList =
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java
index 7de06376b6d..ebb3d90bc8c 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java
@@ -148,7 +148,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SIN
@TreeModel
@TableModel
-public abstract class IoTDBSink implements PipeConnector {
+public abstract class IoTDBSink implements PipeConnector, PipeConnectorWithEventDiscard {
private static final String PARSE_URL_ERROR_FORMATTER =
"Exception occurred while parsing node urls from target servers: {}";
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/PipeConnectorWithEventDiscard.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/PipeConnectorWithEventDiscard.java
new file mode 100644
index 00000000000..ab4dbcf9075
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/PipeConnectorWithEventDiscard.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.pipe.sink.protocol;
+
+public interface PipeConnectorWithEventDiscard {
+
+ void discardEventsOfPipe(String pipeName, long creationTime, int regionId);
+}
(iotdb) 03/04: drop
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch drop-pipe-fix
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit a3b7ea0bcd4173a562ce1f9d50c5728f79171db6
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Mon Apr 27 14:28:36 2026 +0800
drop
---
.../agent/task/connection/PipeEventCollector.java | 18 +++--
.../sink/PipeRealtimePriorityBlockingQueue.java | 4 +-
.../task/connection/PipeEventCollectorTest.java | 87 ++++++++++++++++++++++
.../task/connection/BlockingPendingQueue.java | 40 +++++++++-
4 files changed, 139 insertions(+), 10 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
index a22848ee3ba..b000c5d2366 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
@@ -220,7 +220,8 @@ public class PipeEventCollector implements EventCollector {
private void collectEvent(final Event event) {
if (event instanceof EnrichedEvent) {
- if (!((EnrichedEvent) event).increaseReferenceCount(PipeEventCollector.class.getName())) {
+ final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
+ if (!enrichedEvent.increaseReferenceCount(PipeEventCollector.class.getName())) {
LOGGER.warn("PipeEventCollector: The event {} is already released, skipping it.", event);
isFailedToIncreaseReferenceCount = true;
return;
@@ -228,18 +229,25 @@ public class PipeEventCollector implements EventCollector {
// Assign a commit id for this event in order to report progress in order.
PipeEventCommitManager.getInstance()
- .enrichWithCommitterKeyAndCommitId((EnrichedEvent) event, creationTime, regionId);
+ .enrichWithCommitterKeyAndCommitId(enrichedEvent, creationTime, regionId);
// Assign a rebootTime for iotConsensusV2
- ((EnrichedEvent) event).setRebootTimes(PipeDataNodeAgent.runtime().getRebootTimes());
+ enrichedEvent.setRebootTimes(PipeDataNodeAgent.runtime().getRebootTimes());
+
+ if (enrichedEvent.getPipeName() != null
+ && pendingQueue.isPipeDropped(enrichedEvent.getPipeName(), creationTime, regionId)) {
+ enrichedEvent.clearReferenceCount(PipeEventCollector.class.getName());
+ return;
+ }
}
if (event instanceof PipeHeartbeatEvent) {
((PipeHeartbeatEvent) event).recordConnectorQueueSize(pendingQueue);
}
- pendingQueue.offer(event);
- collectInvocationCount.incrementAndGet();
+ if (pendingQueue.offer(event)) {
+ collectInvocationCount.incrementAndGet();
+ }
}
public void resetFlags() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java
index 3d553f73595..f972bba0e6e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java
@@ -73,7 +73,9 @@ public class PipeRealtimePriorityBlockingQueue extends UnboundedBlockingPendingQ
@Override
public boolean offer(final Event event) {
- checkBeforeOffer(event);
+ if (!checkBeforeOffer(event)) {
+ return false;
+ }
if (event instanceof TsFileInsertionEvent) {
tsfileInsertEventDeque.add((TsFileInsertionEvent) event);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java
new file mode 100644
index 00000000000..bbd5e0b5e3d
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.agent.task.connection;
+
+import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.agent.task.subtask.sink.PipeRealtimePriorityBlockingQueue;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter;
+import org.apache.iotdb.pipe.api.event.Event;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class PipeEventCollectorTest {
+
+ @Test
+ public void testCollectorDoesNotOfferEventsOfDroppedPipeToUnboundedPendingQueue() {
+ verifyCollectorDoesNotOfferEventsOfDroppedPipe(
+ new UnboundedBlockingPendingQueue<>(new PipeDataRegionEventCounter()));
+ }
+
+ @Test
+ public void testCollectorDoesNotOfferEventsOfDroppedPipeToRealtimePendingQueue() {
+ verifyCollectorDoesNotOfferEventsOfDroppedPipe(new PipeRealtimePriorityBlockingQueue());
+ }
+
+ private void verifyCollectorDoesNotOfferEventsOfDroppedPipe(
+ final UnboundedBlockingPendingQueue<Event> pendingQueue) {
+ pendingQueue.discardEventsOfPipe("pipe", 1L, 1);
+
+ final PipeEventCollector droppedPipeCollector =
+ new PipeEventCollector(pendingQueue, 1L, 1, false, false, false);
+ final PipeRawTabletInsertionEvent droppedPipeEvent =
+ createPipeRawTabletInsertionEvent("pipe", 1L);
+ droppedPipeCollector.collect(droppedPipeEvent);
+
+ Assert.assertTrue(droppedPipeEvent.isReleased());
+ Assert.assertEquals(0, pendingQueue.size());
+
+ final PipeEventCollector recreatedPipeCollector =
+ new PipeEventCollector(pendingQueue, 2L, 1, false, false, false);
+ final PipeRawTabletInsertionEvent recreatedPipeEvent =
+ createPipeRawTabletInsertionEvent("pipe", 2L);
+ recreatedPipeCollector.collect(recreatedPipeEvent);
+
+ Assert.assertFalse(recreatedPipeEvent.isReleased());
+ Assert.assertEquals(1, pendingQueue.size());
+
+ pendingQueue.discardAllEvents();
+ Assert.assertTrue(recreatedPipeEvent.isReleased());
+ }
+
+ private PipeRawTabletInsertionEvent createPipeRawTabletInsertionEvent(
+ final String pipeName, final long creationTime) {
+ final List<IMeasurementSchema> schemaList =
+ Arrays.asList(new MeasurementSchema("s1", TSDataType.INT64));
+ final Tablet tablet = new Tablet("root.db.d1", schemaList, 1);
+ tablet.addTimestamp(0, 1L);
+ tablet.addValue("s1", 0, 1L);
+ return new PipeRawTabletInsertionEvent(
+ false, "root.db", "db", "root.db", tablet, false, pipeName, creationTime, null, null, false);
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java
index b3b796ab6d8..7080a2fe6f9 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java
@@ -28,8 +28,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.Set;
import java.util.function.Consumer;
public abstract class BlockingPendingQueue<E extends Event> {
@@ -43,6 +45,7 @@ public abstract class BlockingPendingQueue<E extends Event> {
protected final PipeEventCounter eventCounter;
protected final AtomicBoolean isClosed = new AtomicBoolean(false);
+ protected final Set<String> droppedPipeTaskKeys = ConcurrentHashMap.newKeySet();
protected BlockingPendingQueue(
final BlockingQueue<E> pendingQueue, final PipeEventCounter eventCounter) {
@@ -51,7 +54,10 @@ public abstract class BlockingPendingQueue<E extends Event> {
}
public boolean offer(final E event) {
- checkBeforeOffer(event);
+ if (!checkBeforeOffer(event)) {
+ return false;
+ }
+
final boolean offered = pendingQueue.offer(event);
if (offered) {
eventCounter.increaseEventCount(event);
@@ -60,7 +66,9 @@ public abstract class BlockingPendingQueue<E extends Event> {
}
public boolean put(final E event) {
- checkBeforeOffer(event);
+ if (!checkBeforeOffer(event)) {
+ return false;
+ }
try {
pendingQueue.put(event);
eventCounter.increaseEventCount(event);
@@ -101,6 +109,7 @@ public abstract class BlockingPendingQueue<E extends Event> {
isClosed.set(true);
pendingQueue.clear();
eventCounter.reset();
+ droppedPipeTaskKeys.clear();
}
/** DO NOT FORGET to set eventCounter to new value after invoking this method. */
@@ -120,10 +129,12 @@ public abstract class BlockingPendingQueue<E extends Event> {
return true;
});
eventCounter.reset();
+ droppedPipeTaskKeys.clear();
}
public void discardEventsOfPipe(
final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) {
+ droppedPipeTaskKeys.add(generatePipeTaskKey(pipeNameToDrop, creationTimeToDrop, regionId));
pendingQueue.removeIf(
event -> {
if (event instanceof EnrichedEvent
@@ -158,10 +169,12 @@ public abstract class BlockingPendingQueue<E extends Event> {
return eventCounter.getPipeHeartbeatEventCount();
}
- protected void checkBeforeOffer(final E event) {
- if (isClosed.get() && event instanceof EnrichedEvent) {
+ protected boolean checkBeforeOffer(final E event) {
+ final boolean shouldReject = isClosed.get() || isEventFromDroppedPipe(event);
+ if (shouldReject && event instanceof EnrichedEvent) {
((EnrichedEvent) event).clearReferenceCount(BlockingPendingQueue.class.getName());
}
+ return !shouldReject;
}
protected static boolean isEventFromPipe(
@@ -173,4 +186,23 @@ public abstract class BlockingPendingQueue<E extends Event> {
&& creationTimeToDrop == event.getCreationTime()
&& regionId == event.getRegionId();
}
+
+ protected boolean isEventFromDroppedPipe(final E event) {
+ return event instanceof EnrichedEvent
+ && ((EnrichedEvent) event).getPipeName() != null
+ && isPipeDropped(
+ ((EnrichedEvent) event).getPipeName(),
+ ((EnrichedEvent) event).getCreationTime(),
+ ((EnrichedEvent) event).getRegionId());
+ }
+
+ public boolean isPipeDropped(
+ final String pipeName, final long creationTime, final int regionId) {
+ return droppedPipeTaskKeys.contains(generatePipeTaskKey(pipeName, creationTime, regionId));
+ }
+
+ private static String generatePipeTaskKey(
+ final String pipeName, final long creationTime, final int regionId) {
+ return pipeName + "_" + creationTime + "_" + regionId;
+ }
}
(iotdb) 01/04: drop-1
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch drop-pipe-fix
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 4ec94cd7bf7668a7bbc57ffb14ae68317f8832a6
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Mon Apr 27 11:18:40 2026 +0800
drop-1
---
.../sink/PipeRealtimePriorityBlockingQueue.java | 9 ++--
.../agent/task/subtask/sink/PipeSinkSubtask.java | 10 ++--
.../subtask/sink/PipeSinkSubtaskLifeCycle.java | 5 +-
.../task/subtask/sink/PipeSinkSubtaskManager.java | 2 +-
.../evolvable/batch/PipeTabletEventBatch.java | 7 ++-
.../batch/PipeTransferBatchReqBuilder.java | 8 ++--
.../thrift/async/IoTDBDataRegionAsyncSink.java | 51 ++++++++++++++++----
.../thrift/sync/IoTDBDataRegionSyncSink.java | 5 +-
.../subtask/SubscriptionSinkSubtaskLifeCycle.java | 5 +-
.../subtask/SubscriptionSinkSubtaskManager.java | 2 +-
.../apache/iotdb/db/pipe/sink/PipeSinkTest.java | 53 +++++++++++++++++++++
.../opc_security/8443_12686/iotdb-server.pfx | Bin 0 -> 2942 bytes
.../task/connection/BlockingPendingQueue.java | 17 +++++--
.../commons/pipe/sink/protocol/IoTDBSink.java | 3 +-
14 files changed, 145 insertions(+), 32 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java
index 6d227ac31fd..3d553f73595 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java
@@ -356,13 +356,14 @@ public class PipeRealtimePriorityBlockingQueue extends UnboundedBlockingPendingQ
}
@Override
- public void discardEventsOfPipe(final String pipeNameToDrop, final int regionId) {
- super.discardEventsOfPipe(pipeNameToDrop, regionId);
+ public void discardEventsOfPipe(
+ final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) {
+ super.discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId);
tsfileInsertEventDeque.removeIf(
event -> {
if (event instanceof EnrichedEvent
- && pipeNameToDrop.equals(((EnrichedEvent) event).getPipeName())
- && regionId == ((EnrichedEvent) event).getRegionId()) {
+ && isEventFromPipe(
+ ((EnrichedEvent) event), pipeNameToDrop, creationTimeToDrop, regionId)) {
if (((EnrichedEvent) event)
.clearReferenceCount(PipeRealtimePriorityBlockingQueue.class.getName())) {
eventCounter.decreaseEventCount(event);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
index 6e26a76d774..f6008822e61 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
@@ -198,9 +198,10 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
* When a pipe is dropped, the connector maybe reused and will not be closed. So we just discard
* its queued events in the output pipe connector.
*/
- public void discardEventsOfPipe(final String pipeNameToDrop, int regionId) {
+ public void discardEventsOfPipe(
+ final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) {
// Try to remove the events as much as possible
- inputPendingQueue.discardEventsOfPipe(pipeNameToDrop, regionId);
+ inputPendingQueue.discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId);
try {
increaseHighPriorityTaskCount();
@@ -214,6 +215,7 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
// will.
if (lastEvent instanceof EnrichedEvent
&& pipeNameToDrop.equals(((EnrichedEvent) lastEvent).getPipeName())
+ && creationTimeToDrop == ((EnrichedEvent) lastEvent).getCreationTime()
&& regionId == ((EnrichedEvent) lastEvent).getRegionId()) {
// Do not clear the last event's reference counts because it may be on transferring
lastEvent = null;
@@ -237,6 +239,7 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
// "nonnull" detection.
if (lastExceptionEvent instanceof EnrichedEvent
&& pipeNameToDrop.equals(((EnrichedEvent) lastExceptionEvent).getPipeName())
+ && creationTimeToDrop == ((EnrichedEvent) lastExceptionEvent).getCreationTime()
&& regionId == ((EnrichedEvent) lastExceptionEvent).getRegionId()) {
clearReferenceCountAndReleaseLastExceptionEvent();
}
@@ -246,7 +249,8 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
}
if (outputPipeSink instanceof IoTDBSink) {
- ((IoTDBSink) outputPipeSink).discardEventsOfPipe(pipeNameToDrop, regionId);
+ ((IoTDBSink) outputPipeSink)
+ .discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskLifeCycle.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskLifeCycle.java
index 35f7983075d..1780f5a87ef 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskLifeCycle.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskLifeCycle.java
@@ -92,12 +92,13 @@ public class PipeSinkSubtaskLifeCycle implements AutoCloseable {
* {@link PipeSinkSubtask} should never be used again
* @throws IllegalStateException if {@link PipeSinkSubtaskLifeCycle#registeredTaskCount} <= 0
*/
- public synchronized boolean deregister(final String pipeNameToDeregister, int regionId) {
+ public synchronized boolean deregister(
+ final String pipeNameToDeregister, final long creationTimeToDeregister, final int regionId) {
if (registeredTaskCount <= 0) {
throw new IllegalStateException("registeredTaskCount <= 0");
}
- subtask.discardEventsOfPipe(pipeNameToDeregister, regionId);
+ subtask.discardEventsOfPipe(pipeNameToDeregister, creationTimeToDeregister, regionId);
try {
if (registeredTaskCount > 1) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
index 9138a075918..4faa42db004 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
@@ -209,7 +209,7 @@ public class PipeSinkSubtaskManager {
// Shall not be empty
final PipeSinkSubtaskExecutor executor = lifeCycles.get(0).executor;
- lifeCycles.removeIf(o -> o.deregister(pipeName, regionId));
+ lifeCycles.removeIf(o -> o.deregister(pipeName, creationTime, regionId));
if (lifeCycles.isEmpty()) {
attributeSortedString2SubtaskLifeCycleMap.remove(attributeSortedString);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
index 96bddd0d672..c44e12a4bbf 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
@@ -154,10 +154,13 @@ public abstract class PipeTabletEventBatch implements AutoCloseable {
* Discard all events of the given pipe. This method only clears the reference count of the events
* and discard them, but do not modify other objects (such as buffers) for simplicity.
*/
- public synchronized void discardEventsOfPipe(final String pipeNameToDrop, final int regionId) {
+ public synchronized void discardEventsOfPipe(
+ final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) {
events.removeIf(
event -> {
- if (pipeNameToDrop.equals(event.getPipeName()) && regionId == event.getRegionId()) {
+ if (pipeNameToDrop.equals(event.getPipeName())
+ && creationTimeToDrop == event.getCreationTime()
+ && regionId == event.getRegionId()) {
event.clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
return true;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java
index dd4d4fe1ce6..5bb76ae40e5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java
@@ -196,9 +196,11 @@ public class PipeTransferBatchReqBuilder implements AutoCloseable {
&& endPointToBatch.values().stream().allMatch(PipeTabletEventPlainBatch::isEmpty);
}
- public synchronized void discardEventsOfPipe(final String pipeNameToDrop, final int regionId) {
- defaultBatch.discardEventsOfPipe(pipeNameToDrop, regionId);
- endPointToBatch.values().forEach(batch -> batch.discardEventsOfPipe(pipeNameToDrop, regionId));
+ public synchronized void discardEventsOfPipe(
+ final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) {
+ defaultBatch.discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId);
+ endPointToBatch.values().forEach(
+ batch -> batch.discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId));
}
public int size() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
index d19639310cf..c00e36e0c7c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
@@ -77,6 +77,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -126,6 +127,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
private final AtomicBoolean isClosed = new AtomicBoolean(false);
private final Map<PipeTransferTrackableHandler, PipeTransferTrackableHandler> pendingHandlers =
new ConcurrentHashMap<>();
+ private final Set<String> droppedPipeTaskKeys = ConcurrentHashMap.newKeySet();
private boolean enableSendTsFileLimit;
private volatile boolean isConnectionException;
@@ -681,8 +683,15 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
public void addFailureEventToRetryQueue(final Event event, final Exception e) {
isConnectionException =
e instanceof PipeConnectionException || ThriftClient.isConnectionBroken(e);
- if (event instanceof EnrichedEvent && ((EnrichedEvent) event).isReleased()) {
- return;
+ if (event instanceof EnrichedEvent) {
+ final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
+ if (enrichedEvent.isReleased()) {
+ return;
+ }
+ if (isDroppedPipe(enrichedEvent)) {
+ enrichedEvent.clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
+ return;
+ }
}
if (isClosed.get()) {
@@ -728,15 +737,18 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
//////////////////////////// Operations for close ////////////////////////////
@Override
- public synchronized void discardEventsOfPipe(final String pipeNameToDrop, final int regionId) {
- if (isTabletBatchModeEnabled) {
- tabletBatchBuilder.discardEventsOfPipe(pipeNameToDrop, regionId);
+ public synchronized void discardEventsOfPipe(
+ final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) {
+ droppedPipeTaskKeys.add(generatePipeTaskKey(pipeNameToDrop, creationTimeToDrop, regionId));
+
+ if (isTabletBatchModeEnabled && Objects.nonNull(tabletBatchBuilder)) {
+ tabletBatchBuilder.discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId);
}
retryEventQueue.removeIf(
event -> {
if (event instanceof EnrichedEvent
- && pipeNameToDrop.equals(((EnrichedEvent) event).getPipeName())
- && regionId == ((EnrichedEvent) event).getRegionId()) {
+ && isDroppedPipe(
+ (EnrichedEvent) event, pipeNameToDrop, creationTimeToDrop, regionId)) {
((EnrichedEvent) event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
retryEventQueueEventCounter.decreaseEventCount(event);
return true;
@@ -747,8 +759,8 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
retryTsFileQueue.removeIf(
event -> {
if (event instanceof EnrichedEvent
- && pipeNameToDrop.equals(((EnrichedEvent) event).getPipeName())
- && regionId == ((EnrichedEvent) event).getRegionId()) {
+ && isDroppedPipe(
+ (EnrichedEvent) event, pipeNameToDrop, creationTimeToDrop, regionId)) {
((EnrichedEvent) event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
retryEventQueueEventCounter.decreaseEventCount(event);
return true;
@@ -792,6 +804,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
// clear reference count of events in retry queue after closing async client
clearRetryEventsReferenceCount();
+ droppedPipeTaskKeys.clear();
super.close();
}
@@ -848,6 +861,26 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
this.transferTsFileCounter = transferTsFileCounter;
}
+ private boolean isDroppedPipe(final EnrichedEvent event) {
+ return droppedPipeTaskKeys.contains(
+ generatePipeTaskKey(event.getPipeName(), event.getCreationTime(), event.getRegionId()));
+ }
+
+ private static boolean isDroppedPipe(
+ final EnrichedEvent event,
+ final String pipeNameToDrop,
+ final long creationTimeToDrop,
+ final int regionId) {
+ return pipeNameToDrop.equals(event.getPipeName())
+ && creationTimeToDrop == event.getCreationTime()
+ && regionId == event.getRegionId();
+ }
+
+ private static String generatePipeTaskKey(
+ final String pipeName, final long creationTime, final int regionId) {
+ return pipeName + "_" + creationTime + "_" + regionId;
+ }
+
@Override
public void setTabletBatchSizeHistogram(Histogram tabletBatchSizeHistogram) {
if (tabletBatchBuilder != null) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
index 9357a8c6a6d..e8c4420861c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
@@ -599,9 +599,10 @@ public class IoTDBDataRegionSyncSink extends IoTDBDataNodeSyncSink {
}
@Override
- public synchronized void discardEventsOfPipe(final String pipeNameToDrop, final int regionId) {
+ public synchronized void discardEventsOfPipe(
+ final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) {
if (Objects.nonNull(tabletBatchBuilder)) {
- tabletBatchBuilder.discardEventsOfPipe(pipeNameToDrop, regionId);
+ tabletBatchBuilder.discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java
index 98163697374..390a6d58018 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java
@@ -63,7 +63,10 @@ public class SubscriptionSinkSubtaskLifeCycle extends PipeSinkSubtaskLifeCycle {
}
@Override
- public synchronized boolean deregister(final String pipeNameToDeregister, int regionId) {
+ public synchronized boolean deregister(
+ final String pipeNameToDeregister,
+ final long creationTimeToDeregister,
+ final int regionId) {
if (registeredTaskCount <= 0) {
throw new IllegalStateException("registeredTaskCount <= 0");
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskManager.java
index 6d5f27d8172..16a9ee1a03d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskManager.java
@@ -168,7 +168,7 @@ public class SubscriptionSinkSubtaskManager {
final PipeSinkSubtaskLifeCycle lifeCycle =
attributeSortedString2SubtaskLifeCycleMap.get(attributeSortedString);
- if (lifeCycle.deregister(pipeName, regionId)) {
+ if (lifeCycle.deregister(pipeName, creationTime, regionId)) {
attributeSortedString2SubtaskLifeCycleMap.remove(attributeSortedString);
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
index ec2122b9175..7dfd0446038 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.pipe.sink;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSinkRuntimeEnvironment;
@@ -30,6 +31,7 @@ import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncS
import org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.write.record.Tablet;
@@ -104,6 +106,46 @@ public class PipeSinkTest {
}
}
+ @Test
+ public void testAsyncSinkDropDoesNotRequeueDroppedPipeEvents() throws Exception {
+ try (final IoTDBDataRegionAsyncSink connector = new IoTDBDataRegionAsyncSink()) {
+ final PipeParameters parameters =
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSinkConstant.CONNECTOR_KEY,
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName());
+ put(PipeSinkConstant.CONNECTOR_IOTDB_NODE_URLS_KEY, "127.0.0.1:6668");
+ }
+ });
+ connector.validate(new PipeParameterValidator(parameters));
+ connector.customize(
+ parameters,
+ new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("pipe", 1L, 1)));
+
+ final PipeRawTabletInsertionEvent droppedEvent =
+ createPipeRawTabletInsertionEvent("pipe", 1L, 1);
+ droppedEvent.increaseReferenceCount("test");
+ droppedEvent.setCommitterKeyAndCommitId(new CommitterKey("pipe", 1L, 1, -1), 1L);
+
+ connector.discardEventsOfPipe("pipe", 1L, 1);
+ connector.addFailureEventToRetryQueue(droppedEvent, new PipeException("test"));
+
+ Assert.assertEquals(0, connector.getRetryEventQueueSize());
+ Assert.assertTrue(droppedEvent.isReleased());
+
+ final PipeRawTabletInsertionEvent recreatedPipeEvent =
+ createPipeRawTabletInsertionEvent("pipe", 2L, 1);
+ recreatedPipeEvent.increaseReferenceCount("test");
+ recreatedPipeEvent.setCommitterKeyAndCommitId(new CommitterKey("pipe", 2L, 1, -1), 1L);
+
+ connector.addFailureEventToRetryQueue(recreatedPipeEvent, new PipeException("test"));
+
+ Assert.assertEquals(1, connector.getRetryEventQueueSize());
+ }
+ }
+
@Test
public void testOpcUaSink() {
final List<IMeasurementSchema> schemaList =
@@ -194,4 +236,15 @@ public class PipeSinkTest {
Assert.fail();
}
}
+
+ private PipeRawTabletInsertionEvent createPipeRawTabletInsertionEvent(
+ final String pipeName, final long creationTime, final int regionId) {
+ final List<IMeasurementSchema> schemaList =
+ Arrays.asList(new MeasurementSchema("s1", TSDataType.INT64));
+ final Tablet tablet = new Tablet("root.db.d" + regionId, schemaList, 1);
+ tablet.addTimestamp(0, 1L);
+ tablet.addValue("s1", 0, 1L);
+ return new PipeRawTabletInsertionEvent(
+ false, "root.db", "db", "root.db", tablet, false, pipeName, creationTime, null, null, false);
+ }
}
diff --git a/iotdb-core/datanode/src/test/resources/opc_security/8443_12686/iotdb-server.pfx b/iotdb-core/datanode/src/test/resources/opc_security/8443_12686/iotdb-server.pfx
new file mode 100644
index 00000000000..b36d23da368
Binary files /dev/null and b/iotdb-core/datanode/src/test/resources/opc_security/8443_12686/iotdb-server.pfx differ
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java
index 8773b03f9f3..b3b796ab6d8 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java
@@ -122,12 +122,13 @@ public abstract class BlockingPendingQueue<E extends Event> {
eventCounter.reset();
}
- public void discardEventsOfPipe(final String pipeNameToDrop, final int regionId) {
+ public void discardEventsOfPipe(
+ final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) {
pendingQueue.removeIf(
event -> {
if (event instanceof EnrichedEvent
- && pipeNameToDrop.equals(((EnrichedEvent) event).getPipeName())
- && regionId == ((EnrichedEvent) event).getRegionId()) {
+ && isEventFromPipe(
+ ((EnrichedEvent) event), pipeNameToDrop, creationTimeToDrop, regionId)) {
if (((EnrichedEvent) event).clearReferenceCount(BlockingPendingQueue.class.getName())) {
eventCounter.decreaseEventCount(event);
}
@@ -162,4 +163,14 @@ public abstract class BlockingPendingQueue<E extends Event> {
((EnrichedEvent) event).clearReferenceCount(BlockingPendingQueue.class.getName());
}
}
+
+ protected static boolean isEventFromPipe(
+ final EnrichedEvent event,
+ final String pipeNameToDrop,
+ final long creationTimeToDrop,
+ final int regionId) {
+ return pipeNameToDrop.equals(event.getPipeName())
+ && creationTimeToDrop == event.getCreationTime()
+ && regionId == event.getRegionId();
+ }
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java
index 88a8b71775f..7de06376b6d 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java
@@ -641,7 +641,8 @@ public abstract class IoTDBSink implements PipeConnector {
* When a pipe is dropped, the connector maybe reused and will not be closed. We need to discard
* its batched or queued events in the output pipe connector.
*/
- public synchronized void discardEventsOfPipe(final String pipeName, final int regionId) {
+ public synchronized void discardEventsOfPipe(
+ final String pipeName, final long creationTime, final int regionId) {
// Do nothing by default
}
(iotdb) 04/04: fix
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch drop-pipe-fix
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d8114969d0a7601c3697f09cfedc74523c8fcfe0
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Mon Apr 27 18:50:05 2026 +0800
fix
---
.../db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java | 2 +-
.../evolvable/batch/PipeTransferBatchReqBuilder.java | 5 +++--
.../sink/protocol/websocket/WebSocketConnectorServer.java | 13 +++++++------
.../db/pipe/sink/protocol/websocket/WebSocketSink.java | 2 +-
.../task/subtask/SubscriptionSinkSubtaskLifeCycle.java | 4 +---
.../pipe/agent/task/connection/PipeEventCollectorTest.java | 12 +++++++++++-
.../java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java | 12 +++++++++++-
.../pipe/agent/task/connection/BlockingPendingQueue.java | 5 ++---
8 files changed, 37 insertions(+), 18 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
index dd31d5b5fd8..a11a1a68f0c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
@@ -25,8 +25,8 @@ import org.apache.iotdb.commons.pipe.agent.task.subtask.PipeAbstractSinkSubtask;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSink;
-import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.commons.pipe.sink.protocol.PipeConnectorWithEventDiscard;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.event.UserDefinedEnrichedEvent;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java
index 5bb76ae40e5..49d9d8cea09 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java
@@ -199,8 +199,9 @@ public class PipeTransferBatchReqBuilder implements AutoCloseable {
public synchronized void discardEventsOfPipe(
final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) {
defaultBatch.discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId);
- endPointToBatch.values().forEach(
- batch -> batch.discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId));
+ endPointToBatch
+ .values()
+ .forEach(batch -> batch.discardEventsOfPipe(pipeNameToDrop, creationTimeToDrop, regionId));
}
public int size() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketConnectorServer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketConnectorServer.java
index 6b402b27026..4706eb0275a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketConnectorServer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketConnectorServer.java
@@ -125,8 +125,7 @@ public class WebSocketConnectorServer extends WebSocketServer {
if (eventTransferQueue != null) {
eventTransferQueue.removeIf(
eventWrapper ->
- discardIfMatches(
- eventWrapper.event, pipeNameToDrop, creationTimeToDrop, regionId));
+ discardIfMatches(eventWrapper.event, pipeNameToDrop, creationTimeToDrop, regionId));
synchronized (eventTransferQueue) {
eventTransferQueue.notifyAll();
}
@@ -135,10 +134,12 @@ public class WebSocketConnectorServer extends WebSocketServer {
final ConcurrentHashMap<Long, EventWaitingForAck> eventId2EventMap =
eventsWaitingForAck.get(pipeNameToDrop);
if (eventId2EventMap != null) {
- eventId2EventMap.entrySet().removeIf(
- entry ->
- discardIfMatches(
- entry.getValue().event, pipeNameToDrop, creationTimeToDrop, regionId));
+ eventId2EventMap
+ .entrySet()
+ .removeIf(
+ entry ->
+ discardIfMatches(
+ entry.getValue().event, pipeNameToDrop, creationTimeToDrop, regionId));
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java
index dadeee8053d..e8a38d63e6e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java
@@ -19,9 +19,9 @@
package org.apache.iotdb.db.pipe.sink.protocol.websocket;
-import org.apache.iotdb.commons.pipe.sink.protocol.PipeConnectorWithEventDiscard;
import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.commons.pipe.sink.protocol.PipeConnectorWithEventDiscard;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java
index 390a6d58018..af871feaa7e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskLifeCycle.java
@@ -64,9 +64,7 @@ public class SubscriptionSinkSubtaskLifeCycle extends PipeSinkSubtaskLifeCycle {
@Override
public synchronized boolean deregister(
- final String pipeNameToDeregister,
- final long creationTimeToDeregister,
- final int regionId) {
+ final String pipeNameToDeregister, final long creationTimeToDeregister, final int regionId) {
if (registeredTaskCount <= 0) {
throw new IllegalStateException("registeredTaskCount <= 0");
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java
index bbd5e0b5e3d..029a722c8a9 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java
@@ -82,6 +82,16 @@ public class PipeEventCollectorTest {
tablet.addTimestamp(0, 1L);
tablet.addValue("s1", 0, 1L);
return new PipeRawTabletInsertionEvent(
- false, "root.db", "db", "root.db", tablet, false, pipeName, creationTime, null, null, false);
+ false,
+ "root.db",
+ "db",
+ "root.db",
+ tablet,
+ false,
+ pipeName,
+ creationTime,
+ null,
+ null,
+ false);
}
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
index 3ad262130ff..cf311639ee9 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
@@ -287,6 +287,16 @@ public class PipeSinkTest {
tablet.addTimestamp(0, 1L);
tablet.addValue("s1", 0, 1L);
return new PipeRawTabletInsertionEvent(
- false, "root.db", "db", "root.db", tablet, false, pipeName, creationTime, null, null, false);
+ false,
+ "root.db",
+ "db",
+ "root.db",
+ tablet,
+ false,
+ pipeName,
+ creationTime,
+ null,
+ null,
+ false);
}
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java
index 7080a2fe6f9..adbc79d5004 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java
@@ -27,11 +27,11 @@ import org.apache.iotdb.pipe.api.event.Event;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.Set;
import java.util.function.Consumer;
public abstract class BlockingPendingQueue<E extends Event> {
@@ -196,8 +196,7 @@ public abstract class BlockingPendingQueue<E extends Event> {
((EnrichedEvent) event).getRegionId());
}
- public boolean isPipeDropped(
- final String pipeName, final long creationTime, final int regionId) {
+ public boolean isPipeDropped(final String pipeName, final long creationTime, final int regionId) {
return droppedPipeTaskKeys.contains(generatePipeTaskKey(pipeName, creationTime, regionId));
}