This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 59ac348a3f2 Fix Pipe leader cache isolation across receivers (#19064)
59ac348a3f2 is described below
commit 59ac348a3f2954cbdbc5e53c01a3cd475a187880
Author: Caideyipi <[email protected]>
AuthorDate: Mon Nov 2 10:29:40 2026 +0800
Fix Pipe leader cache isolation across receivers (#19064)
---
.../pipe/it/triple/AbstractPipeTripleManualIT.java | 6 +-
.../iotdb/pipe/it/triple/IoTDBPipeForwardIT.java | 84 +++++++++++++++++++++
.../client/IoTDBDataNodeAsyncClientManager.java | 11 ++-
.../IoTDBDataNodeCacheLeaderClientManager.java | 36 +++++++--
.../client/IoTDBDataNodeSyncClientManager.java | 12 ++-
.../batch/PipeTransferBatchReqBuilder.java | 13 +++-
.../thrift/async/IoTDBDataRegionAsyncSink.java | 3 +-
.../thrift/sync/IoTDBDataRegionSyncSink.java | 3 +-
.../sink/IoTDBDataNodeAsyncClientManagerTest.java | 86 ++++++++++++++++++----
.../batch/PipeTransferBatchReqBuilderTest.java | 55 ++++++++++++++
10 files changed, 279 insertions(+), 30 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/triple/AbstractPipeTripleManualIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/triple/AbstractPipeTripleManualIT.java
index c1d92c7916b..0640f3fb645 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/triple/AbstractPipeTripleManualIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/triple/AbstractPipeTripleManualIT.java
@@ -39,11 +39,15 @@ abstract class AbstractPipeTripleManualIT {
env2 = MultiEnvFactory.getEnv(1);
env3 = MultiEnvFactory.getEnv(2);
setupConfig();
- env1.initClusterEnvironment(1, 1);
+ env1.initClusterEnvironment(1, getEnv1DataNodeCount());
env2.initClusterEnvironment(1, 1);
env3.initClusterEnvironment(1, 1);
}
+ protected int getEnv1DataNodeCount() {
+ return 1;
+ }
+
protected void setupConfig() {
env1.getConfig()
.getCommonConfig()
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/triple/IoTDBPipeForwardIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/triple/IoTDBPipeForwardIT.java
index 3f2cab34501..6beec673feb 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/triple/IoTDBPipeForwardIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/triple/IoTDBPipeForwardIT.java
@@ -22,25 +22,109 @@ package org.apache.iotdb.pipe.it.triple;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.isession.ISession;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.MultiClusterIT3;
+import org.apache.iotdb.itbase.env.BaseEnv;
import org.apache.iotdb.rpc.TSStatusCode;
+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 org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
@RunWith(IoTDBTestRunner.class)
@Category({MultiClusterIT3.class})
public class IoTDBPipeForwardIT extends AbstractPipeTripleManualIT {
+
+ @Override
+ protected int getEnv1DataNodeCount() {
+ return 3;
+ }
+
+ @Override
+ protected void setupConfig() {
+ super.setupConfig();
+ for (final BaseEnv env : Arrays.asList(env1, env2, env3)) {
+ env.getConfig()
+ .getCommonConfig()
+ .setPipeAutoSplitFullEnabled(true)
+ .setAutoCreateSchemaEnabled(true)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS);
+ }
+ }
+
+ @Test
+ public void
testLocalWriteIsForwardedToDownstreamWithDefaultSourceParameters() throws
Exception {
+ TestUtils.executeNonQueries(
+ env1,
+ Collections.singletonList(
+ "create pipe a2b with source('forwarding-pipe-requests'='false',"
+ + "'capture.tree'='true','capture.table'='true') with
sink('node-urls'='"
+ + env2.getDataNodeWrapper(0).getIpAndPortString()
+ + "')"),
+ null);
+ TestUtils.executeNonQueries(
+ env2,
+ Arrays.asList(
+ "create pipe b2a with source('forwarding-pipe-requests'='false',"
+ + "'capture.tree'='true','capture.table'='true') with
sink('node-urls'='"
+ + env1.getDataNodeWrapperList().stream()
+ .map(DataNodeWrapper::getIpAndPortString)
+ .collect(Collectors.joining(","))
+ + "')",
+ "create pipe b2c with sink('node-urls'='"
+ + env3.getDataNodeWrapper(0).getIpAndPortString()
+ + "')"),
+ null);
+
+ final List<IMeasurementSchema> schemas = new ArrayList<>();
+ for (int sensor = 0; sensor < 10; sensor++) {
+ schemas.add(new MeasurementSchema("s_" + sensor, TSDataType.INT32));
+ }
+ try (final ISession session = env2.getSessionConnection()) {
+ for (int batch = 0; batch < 60; batch++) {
+ for (int device = 0; device < 2; device++) {
+ final Tablet tablet = new Tablet("root.test.sg_0.d_" + device,
schemas, 1000);
+ for (int row = 0; row < 1000; row++) {
+ final long timestamp = 1577808000000L + (batch * 1000L + row) *
500;
+ tablet.addTimestamp(row, timestamp);
+ for (int sensor = 0; sensor < schemas.size(); sensor++) {
+ tablet.addValue("s_" + sensor, row, batch * 1000 + row);
+ }
+ }
+ session.insertAlignedTablet(tablet);
+ }
+ // Pace the writes so the workload continues after the source's
initial flush.
+ Thread.sleep(500);
+ }
+ }
+ for (final BaseEnv env : Arrays.asList(env1, env2, env3)) {
+ TestUtils.assertDataEventuallyOnEnv(
+ env,
+ "select count(s_0) from root.test.sg_0.* align by device",
+ "Device,count(s_0),",
+ new HashSet<>(Arrays.asList("root.test.sg_0.d_0,60000,",
"root.test.sg_0.d_1,60000,")),
+ 60);
+ }
+ }
+
@Test
public void testForwardingPipeRequests() throws Exception {
final DataNodeWrapper env2DataNode = env2.getDataNodeWrapper(0);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
index 2b85ac51814..cec0c5cda24 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
@@ -76,6 +76,7 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
private static final Map<String, Integer> CLIENT_RESOURCE_REF_COUNT = new
ConcurrentHashMap<>();
private final String receiverAttributes;
private final String clientResourceKey;
+ private final String receiverKey;
// clientResourceKey -> IClientManager<TEndPoint,
AsyncPipeDataTransferServiceClient>
private static final Map<String, IClientManager<TEndPoint,
AsyncPipeDataTransferServiceClient>>
@@ -119,6 +120,7 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
skipIfNoPrivileges);
endPointSet = new HashSet<>(endPoints);
+ receiverKey =
IoTDBDataNodeCacheLeaderClientManager.generateReceiverKey(endPoints);
receiverAttributes =
String.format(
@@ -187,7 +189,7 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
return borrowClient();
}
- return borrowClient(LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId));
+ return borrowClient(getLeaderEndPoint(deviceId));
}
public AsyncPipeDataTransferServiceClient borrowClient(final TEndPoint
endPoint)
@@ -405,6 +407,11 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
}
}
+ @Override
+ public TEndPoint getLeaderEndPoint(final String deviceId) {
+ return useLeaderCache ?
LEADER_CACHE_MANAGER.getLeaderEndPoint(receiverKey, deviceId) : null;
+ }
+
public void updateLeaderCache(final String deviceId, final TEndPoint
endPoint) {
if (!useLeaderCache
|| deviceId == null
@@ -418,7 +425,7 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
endPointSet.add(endPoint);
}
- LEADER_CACHE_MANAGER.updateLeaderEndPoint(deviceId, endPoint);
+ LEADER_CACHE_MANAGER.updateLeaderEndPoint(receiverKey, deviceId, endPoint);
}
public ExecutorService getExecutor() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java
index 8ea2131395a..be471eed96a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java
@@ -25,26 +25,42 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory;
+import org.apache.iotdb.rpc.UrlUtils;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Weigher;
+import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
public interface IoTDBDataNodeCacheLeaderClientManager {
LeaderCacheManager LEADER_CACHE_MANAGER = new LeaderCacheManager();
+ TEndPoint getLeaderEndPoint(String deviceId);
+
+ static String generateReceiverKey(final List<TEndPoint> endPoints) {
+ return endPoints.stream()
+ .map(UrlUtils::convertTEndPointIpv4AndIpv6Url)
+ .distinct()
+ .sorted()
+ .collect(Collectors.joining(",", "[", "]"));
+ }
+
class LeaderCacheManager {
private static final Logger LOGGER =
LoggerFactory.getLogger(LeaderCacheManager.class);
private static final PipeConfig CONFIG = PipeConfig.getInstance();
// leader cache built by LRU
- private final Cache<String, TEndPoint> device2endpoint;
+ // A device can have different leaders in different receiving clusters.
Keep one bounded
+ // cache, but namespace its entries by the sink's configured endpoints,
before redirection.
+ private final Cache<Pair<String, String>, TEndPoint> device2endpoint;
// a hashmap to reuse the created endpoint
private final ConcurrentHashMap<TEndPoint, TEndPoint> endPoints = new
ConcurrentHashMap<>();
@@ -64,7 +80,10 @@ public interface IoTDBDataNodeCacheLeaderClientManager {
device2endpoint =
Caffeine.newBuilder()
.maximumWeight(allocatedMemoryBlock.getMemoryUsageInBytes())
- .weigher((Weigher<String, TEndPoint>) (device, endPoint) ->
device.getBytes().length)
+ .weigher(
+ (Weigher<Pair<String, String>, TEndPoint>)
+ (key, endPoint) ->
+ key.getLeft().getBytes().length +
key.getRight().getBytes().length)
.recordStats()
.build();
@@ -103,20 +122,23 @@ public interface IoTDBDataNodeCacheLeaderClientManager {
});
}
- public TEndPoint getLeaderEndPoint(final String deviceId) {
- return deviceId == null ? null : device2endpoint.getIfPresent(deviceId);
+ public TEndPoint getLeaderEndPoint(final String receiverKey, final String
deviceId) {
+ return deviceId == null
+ ? null
+ : device2endpoint.getIfPresent(new Pair<>(receiverKey, deviceId));
}
- public void updateLeaderEndPoint(final String deviceId, final TEndPoint
endPoint) {
+ public void updateLeaderEndPoint(
+ final String receiverKey, final String deviceId, final TEndPoint
endPoint) {
if (deviceId == null || endPoint == null) {
return;
}
final TEndPoint endPointFromMap = endPoints.putIfAbsent(endPoint,
endPoint);
if (endPointFromMap != null) {
- device2endpoint.put(deviceId, endPointFromMap);
+ device2endpoint.put(new Pair<>(receiverKey, deviceId),
endPointFromMap);
} else {
- device2endpoint.put(deviceId, endPoint);
+ device2endpoint.put(new Pair<>(receiverKey, deviceId), endPoint);
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
index e8f081cff00..024b153c9a7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
@@ -45,6 +45,8 @@ public class IoTDBDataNodeSyncClientManager extends
IoTDBSyncClientManager
private static final Logger LOGGER =
LoggerFactory.getLogger(IoTDBDataNodeSyncClientManager.class);
+ private final String receiverKey;
+
public IoTDBDataNodeSyncClientManager(
final List<TEndPoint> endPoints,
final boolean useSSL,
@@ -79,6 +81,7 @@ public class IoTDBDataNodeSyncClientManager extends
IoTDBSyncClientManager
validateTsFile,
shouldMarkAsPipeRequest,
skipIfNoPrivileges);
+ receiverKey =
IoTDBDataNodeCacheLeaderClientManager.generateReceiverKey(endPoints);
}
@Override
@@ -99,7 +102,7 @@ public class IoTDBDataNodeSyncClientManager extends
IoTDBSyncClientManager
}
public Pair<IoTDBSyncClient, Boolean> getClient(final String deviceId) {
- final TEndPoint endPoint =
LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId);
+ final TEndPoint endPoint = getLeaderEndPoint(deviceId);
return useLeaderCache
&& endPoint != null
&& endPoint2ClientAndStatus.containsKey(endPoint)
@@ -119,6 +122,11 @@ public class IoTDBDataNodeSyncClientManager extends
IoTDBSyncClientManager
: getClient();
}
+ @Override
+ public TEndPoint getLeaderEndPoint(final String deviceId) {
+ return useLeaderCache ?
LEADER_CACHE_MANAGER.getLeaderEndPoint(receiverKey, deviceId) : null;
+ }
+
public void updateLeaderCache(final String deviceId, final TEndPoint
endPoint) {
if (!useLeaderCache
|| deviceId == null
@@ -134,7 +142,7 @@ public class IoTDBDataNodeSyncClientManager extends
IoTDBSyncClientManager
reconstructClient(endPoint);
}
- LEADER_CACHE_MANAGER.updateLeaderEndPoint(deviceId, endPoint);
+ LEADER_CACHE_MANAGER.updateLeaderEndPoint(receiverKey, deviceId,
endPoint);
} catch (final Exception e) {
LOGGER.warn(
DataNodePipeMessages.FAILED_TO_UPDATE_LEADER_CACHE_FOR_DEVICE,
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 45ec6505390..ed12e2cbf91 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
@@ -25,7 +25,6 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
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.sink.client.IoTDBDataNodeCacheLeaderClientManager;
import
org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
import org.apache.iotdb.metrics.impl.DoNothingHistogram;
import org.apache.iotdb.metrics.type.Histogram;
@@ -44,6 +43,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.function.Function;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_FORMAT_HYBRID_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_FORMAT_KEY;
@@ -69,6 +69,7 @@ public class PipeTransferBatchReqBuilder implements
AutoCloseable {
private final boolean usingTsFileBatch;
private final boolean useLeaderCache;
+ private final Function<String, TEndPoint> leaderEndPointResolver;
private final int requestMaxDelayInMs;
private final long requestMaxBatchSizeInBytes;
@@ -89,6 +90,13 @@ public class PipeTransferBatchReqBuilder implements
AutoCloseable {
private final Map<TEndPoint, PipeTabletEventPlainBatch> endPointToBatch =
new HashMap<>();
public PipeTransferBatchReqBuilder(final PipeParameters parameters) {
+ // Transports without leader redirection, such as air gap, always use the
default batch.
+ this(parameters, deviceId -> null);
+ }
+
+ public PipeTransferBatchReqBuilder(
+ final PipeParameters parameters, final Function<String, TEndPoint>
leaderEndPointResolver) {
+ this.leaderEndPointResolver = leaderEndPointResolver;
usingTsFileBatch =
parameters
.getStringOrDefault(
@@ -171,8 +179,7 @@ public class PipeTransferBatchReqBuilder implements
AutoCloseable {
return;
}
- final TEndPoint endPoint =
-
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId);
+ final TEndPoint endPoint = leaderEndPointResolver.apply(deviceId);
if (Objects.isNull(endPoint)) {
defaultBatch.onEvent(event);
return;
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 64f25919750..146692d3a9d 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
@@ -248,7 +248,8 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
transferTsFileClientManager.setPipeInfo(pipeName, creationTime);
if (isTabletBatchModeEnabled) {
- tabletBatchBuilder = new PipeTransferBatchReqBuilder(parameters);
+ tabletBatchBuilder =
+ new PipeTransferBatchReqBuilder(parameters,
clientManager::getLeaderEndPoint);
}
enableSendTsFileLimit =
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 444a374992a..dab0d16a0b5 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
@@ -106,7 +106,8 @@ public class IoTDBDataRegionSyncSink extends
IoTDBDataNodeSyncSink {
// tablet batch mode configuration
if (isTabletBatchModeEnabled) {
- tabletBatchBuilder = new PipeTransferBatchReqBuilder(parameters);
+ tabletBatchBuilder =
+ new PipeTransferBatchReqBuilder(parameters,
clientManager::getLeaderEndPoint);
}
enableSendTsFileLimit =
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java
index c26f35ddc77..41b88f12030 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java
@@ -21,12 +21,14 @@ package org.apache.iotdb.db.pipe.sink;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.audit.UserEntity;
+import org.apache.iotdb.commons.client.IClientManager;
+import
org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeAsyncClientManager;
-import
org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeCacheLeaderClientManager;
import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeSyncClientManager;
import org.junit.Assert;
import org.junit.Test;
+import org.mockito.Mockito;
import java.lang.reflect.Field;
import java.util.ArrayList;
@@ -40,6 +42,72 @@ public class IoTDBDataNodeAsyncClientManagerTest {
private static final List<String> WILDCARD_ADDRESSES =
Arrays.asList("0.0.0.0", "::", "[::]", "0:0:0:0:0:0:0:0", "0::0");
+ @Test
+ public void testAsyncBorrowDoesNotUseLeaderFromAnotherReceiver() throws
Exception {
+ final TEndPoint firstReceiver = new TEndPoint("127.0.0.1", 10667);
+ final TEndPoint secondReceiver = new TEndPoint("127.0.0.1", 11667);
+ final TEndPoint firstLeader = new TEndPoint("127.0.0.1", 10668);
+ final IoTDBDataNodeAsyncClientManager firstManager =
+ createAsyncManager(new
ArrayList<>(Collections.singletonList(firstReceiver)));
+ final IoTDBDataNodeAsyncClientManager secondManager =
+ createAsyncManager(new
ArrayList<>(Collections.singletonList(secondReceiver)));
+
+ final IClientManager<TEndPoint, AsyncPipeDataTransferServiceClient>
clients =
+ Mockito.mock(IClientManager.class);
+ final AsyncPipeDataTransferServiceClient client =
+ Mockito.mock(AsyncPipeDataTransferServiceClient.class);
+ Mockito.when(clients.borrowClient(Mockito.any())).thenReturn(client);
+ Mockito.when(client.isHandshakeFinished()).thenReturn(true);
+ final Field clientManagerField =
+
IoTDBDataNodeAsyncClientManager.class.getDeclaredField("endPoint2Client");
+ clientManagerField.setAccessible(true);
+ clientManagerField.set(secondManager, clients);
+
+ try {
+ firstManager.updateLeaderCache("root.test.shared", firstLeader);
+ Assert.assertSame(client,
secondManager.borrowClient("root.test.shared"));
+ Mockito.verify(clients).borrowClient(secondReceiver);
+ Mockito.verify(clients, Mockito.never()).borrowClient(firstLeader);
+ } finally {
+ firstManager.close();
+ secondManager.close();
+ }
+ }
+
+ @Test
+ public void testLeaderCacheIsScopedToConfiguredReceiverEndPoints() {
+ final TEndPoint firstReceiver = new TEndPoint("127.0.0.1", 12667);
+ final TEndPoint secondReceiver = new TEndPoint("127.0.0.1", 12668);
+ final TEndPoint otherReceiver = new TEndPoint("::1", 13667);
+ final TEndPoint firstLeader = new TEndPoint("127.0.0.1", 12669);
+ final TEndPoint otherLeader = new TEndPoint("::1", 13668);
+ final String device = "root.test.scoped";
+ final IoTDBDataNodeAsyncClientManager asyncManager =
+ createAsyncManager(new ArrayList<>(Arrays.asList(firstReceiver,
secondReceiver)));
+ final TestIoTDBDataNodeSyncClientManager sameReceiverManager =
+ new TestIoTDBDataNodeSyncClientManager(
+ new ArrayList<>(Arrays.asList(secondReceiver, firstReceiver,
firstReceiver)));
+ final TestIoTDBDataNodeSyncClientManager otherManager =
+ new TestIoTDBDataNodeSyncClientManager(
+ new ArrayList<>(Collections.singletonList(otherReceiver)));
+
+ try {
+ asyncManager.updateLeaderCache(device, firstLeader);
+ Assert.assertEquals(firstLeader, asyncManager.getLeaderEndPoint(device));
+ Assert.assertEquals(firstLeader,
sameReceiverManager.getLeaderEndPoint(device));
+ Assert.assertNull(otherManager.getLeaderEndPoint(device));
+
+ otherManager.updateLeaderCache(device, otherLeader);
+ Assert.assertEquals(otherLeader, otherManager.getLeaderEndPoint(device));
+ Assert.assertEquals(firstLeader, asyncManager.getLeaderEndPoint(device));
+ Assert.assertEquals(firstLeader,
sameReceiverManager.getLeaderEndPoint(device));
+ } finally {
+ asyncManager.close();
+ sameReceiverManager.close();
+ otherManager.close();
+ }
+ }
+
@Test
public void testReceiverAttributesShouldDifferentiateSkipIfNoPrivileges()
throws Exception {
final IoTDBDataNodeAsyncClientManager managerWithSkipIf =
@@ -134,8 +202,7 @@ public class IoTDBDataNodeAsyncClientManagerTest {
manager.updateLeaderCache(deviceId, new
TEndPoint(WILDCARD_ADDRESSES.get(i), 6667 + i));
Assert.assertEquals(Collections.singletonList(originalEndPoint),
endPoints);
- Assert.assertNull(
-
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId));
+ Assert.assertNull(manager.getLeaderEndPoint(deviceId));
}
final String ipv6DeviceId = "async-ipv6-device";
@@ -143,10 +210,7 @@ public class IoTDBDataNodeAsyncClientManagerTest {
manager.updateLeaderCache(ipv6DeviceId, ipv6EndPoint);
Assert.assertEquals(Arrays.asList(originalEndPoint, ipv6EndPoint),
endPoints);
- Assert.assertEquals(
- ipv6EndPoint,
-
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(
- ipv6DeviceId));
+ Assert.assertEquals(ipv6EndPoint,
manager.getLeaderEndPoint(ipv6DeviceId));
} finally {
manager.close();
}
@@ -166,8 +230,7 @@ public class IoTDBDataNodeAsyncClientManagerTest {
Assert.assertEquals(Collections.singletonList(originalEndPoint),
endPoints);
Assert.assertEquals(0, manager.getReconstructionCount());
- Assert.assertNull(
-
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId));
+ Assert.assertNull(manager.getLeaderEndPoint(deviceId));
}
final String ipv6DeviceId = "sync-ipv6-device";
@@ -176,10 +239,7 @@ public class IoTDBDataNodeAsyncClientManagerTest {
Assert.assertEquals(Arrays.asList(originalEndPoint, ipv6EndPoint),
endPoints);
Assert.assertEquals(1, manager.getReconstructionCount());
- Assert.assertEquals(
- ipv6EndPoint,
-
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(
- ipv6DeviceId));
+ Assert.assertEquals(ipv6EndPoint,
manager.getLeaderEndPoint(ipv6DeviceId));
} finally {
manager.close();
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilderTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilderTest.java
index d3710ae0e38..48c94a712e6 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilderTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilderTest.java
@@ -20,10 +20,12 @@
package org.apache.iotdb.db.pipe.sink.payload.evolvable.batch;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.audit.UserEntity;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
+import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeAsyncClientManager;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
@@ -34,6 +36,7 @@ import org.apache.tsfile.write.record.Tablet;
import org.junit.Assert;
import org.junit.Test;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -41,6 +44,58 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CON
public class PipeTransferBatchReqBuilderTest {
+ @Test
+ public void testBatchesDoNotUseLeaderFromAnotherReceiver() throws Exception {
+ final TEndPoint firstReceiver = new TEndPoint("127.0.0.1", 14667);
+ final TEndPoint secondReceiver = new TEndPoint("127.0.0.1", 15667);
+ final TEndPoint firstLeader = new TEndPoint("127.0.0.1", 14668);
+ final TEndPoint secondLeader = new TEndPoint("127.0.0.1", 15668);
+ final IoTDBDataNodeAsyncClientManager firstManager =
createClientManager(firstReceiver);
+ final IoTDBDataNodeAsyncClientManager secondManager =
createClientManager(secondReceiver);
+ final PipeParameters parameters =
+ new
PipeParameters(Collections.singletonMap(CONNECTOR_IOTDB_BATCH_DELAY_MS_KEY,
"0"));
+ final String deviceId = "root.test.device";
+
+ try (final PipeTransferBatchReqBuilder firstBuilder =
+ new PipeTransferBatchReqBuilder(parameters,
firstManager::getLeaderEndPoint);
+ final PipeTransferBatchReqBuilder secondBuilder =
+ new PipeTransferBatchReqBuilder(parameters,
secondManager::getLeaderEndPoint)) {
+ firstManager.updateLeaderCache(deviceId, firstLeader);
+ firstBuilder.onEvent(createEvent(1));
+ secondBuilder.onEvent(createEvent(2));
+ Assert.assertEquals(
+ firstLeader,
firstBuilder.getAllNonEmptyAndShouldEmitBatches().get(0).getLeft());
+
Assert.assertNull(secondBuilder.getAllNonEmptyAndShouldEmitBatches().get(0).getLeft());
+
+ secondManager.updateLeaderCache(deviceId, secondLeader);
+ secondBuilder.onEvent(createEvent(3));
+ final List<Pair<TEndPoint, PipeTabletEventBatch>> secondBatches =
+ secondBuilder.getAllNonEmptyAndShouldEmitBatches();
+ Assert.assertEquals(2, secondBatches.size());
+ Assert.assertNull(secondBatches.get(0).getLeft());
+ Assert.assertEquals(secondLeader, secondBatches.get(1).getLeft());
+ Assert.assertEquals(firstLeader,
firstManager.getLeaderEndPoint(deviceId));
+ } finally {
+ firstManager.close();
+ secondManager.close();
+ }
+ }
+
+ private static IoTDBDataNodeAsyncClientManager createClientManager(final
TEndPoint receiver) {
+ return new IoTDBDataNodeAsyncClientManager(
+ new ArrayList<>(Collections.singletonList(receiver)),
+ true,
+ "round-robin",
+ new UserEntity(1L, "user", "cli-host"),
+ "password",
+ true,
+ "sync",
+ true,
+ true,
+ false,
+ true);
+ }
+
@Test
public void testDetachedBatchCannotClearSubsequentEvent() throws Exception {
final PipeTransferBatchReqBuilder builder =
