This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 92465ba993 [Fix][Connector-V2] Include MongoDB shard key metadata
(#11599)
92465ba993 is described below
commit 92465ba99349ccbe9943e0bc52170fb95781a7d3
Author: Jast <[email protected]>
AuthorDate: Sun Aug 23 09:13:40 2026 +0800
[Fix][Connector-V2] Include MongoDB shard key metadata (#11599)
Co-authored-by: zhangshenghang <[email protected]>
---
.../cdc/mongodb/config/MongodbSourceConstants.java | 2 +
.../source/splitters/ShardedSplitStrategy.java | 3 +-
.../seatunnel/cdc/mongodb/utils/MongodbUtils.java | 4 +-
.../test/java/mongodb/utils/MongodbUtilsTest.java | 47 ++++++++++++++++++++++
4 files changed, 53 insertions(+), 3 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbSourceConstants.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbSourceConstants.java
index b4315c95d0..9d298039f0 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbSourceConstants.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbSourceConstants.java
@@ -52,6 +52,8 @@ public class MongodbSourceConstants {
public static final String DOCUMENT_KEY = "documentKey";
+ public static final String SHARD_KEY_FIELD = "key";
+
public static final String NS_FIELD = "ns";
public static final String OPERATION_TYPE = "operationType";
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/splitters/ShardedSplitStrategy.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/splitters/ShardedSplitStrategy.java
index 058535f31b..3005df8070 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/splitters/ShardedSplitStrategy.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/splitters/ShardedSplitStrategy.java
@@ -37,6 +37,7 @@ import java.util.List;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.DROPPED_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.MAX_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.MIN_FIELD;
+import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.SHARD_KEY_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.UNAUTHORIZED_ERROR;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbUtils.readChunks;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbUtils.readCollectionMetadata;
@@ -84,7 +85,7 @@ public class ShardedSplitStrategy implements SplitStrategy {
return SampleBucketSplitStrategy.INSTANCE.split(splitContext);
}
- BsonDocument splitKeys = collectionMetadata.getDocument("key");
+ BsonDocument splitKeys =
collectionMetadata.getDocument(SHARD_KEY_FIELD);
SeaTunnelRowType rowType = shardKeysToRowType(splitKeys);
List<SnapshotSplit> snapshotSplits = new ArrayList<>(chunks.size());
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbUtils.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbUtils.java
index a446d748c4..bf53d50e1f 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbUtils.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbUtils.java
@@ -65,7 +65,6 @@ import static
org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated.IL
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.ADD_NS_FIELD_NAME;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.CHANGE_STREAM_FATAL_ERROR;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.COMMAND_SUCCEED_FLAG;
-import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.DOCUMENT_KEY;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.DOES_NOT_EXIST;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.DROPPED_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.ID_FIELD;
@@ -78,6 +77,7 @@ import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.Mongo
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.NS_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.RESUME_TOKEN;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.SHARD_FIELD;
+import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.SHARD_KEY_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.UUID_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.CollectionDiscoveryUtils.ADD_NS_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.CollectionDiscoveryUtils.includeListAsFlatPattern;
@@ -351,7 +351,7 @@ public class MongodbUtils {
return collection
.find(eq(ID_FIELD, collectionId.identifier()))
- .projection(include(ID_FIELD, UUID_FIELD, DROPPED_FIELD,
DOCUMENT_KEY))
+ .projection(include(ID_FIELD, UUID_FIELD, DROPPED_FIELD,
SHARD_KEY_FIELD))
.first();
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/utils/MongodbUtilsTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/utils/MongodbUtilsTest.java
index b441717da6..b9bd629c2b 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/utils/MongodbUtilsTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/utils/MongodbUtilsTest.java
@@ -20,13 +20,60 @@ package mongodb.utils;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbUtils;
+import org.bson.BsonDocument;
+import org.bson.BsonInt32;
+import org.bson.conversions.Bson;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import com.mongodb.MongoClientSettings;
+import com.mongodb.client.FindIterable;
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoDatabase;
+import io.debezium.relational.TableId;
+
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.NS_FIELD;
+import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.SHARD_KEY_FIELD;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
public class MongodbUtilsTest {
+ @Test
+ void testReadCollectionMetadataIncludesShardKey() {
+ MongoClient mongoClient = mock(MongoClient.class);
+ MongoDatabase configDatabase = mock(MongoDatabase.class);
+ MongoCollection<BsonDocument> collections =
mock(MongoCollection.class);
+ FindIterable<BsonDocument> result = mock(FindIterable.class);
+ BsonDocument collectionMetadata =
+ new BsonDocument(SHARD_KEY_FIELD, new BsonDocument("_id", new
BsonInt32(1)));
+
+ when(mongoClient.getDatabase("config")).thenReturn(configDatabase);
+ when(configDatabase.getCollection("collections", BsonDocument.class))
+ .thenReturn(collections);
+ when(collections.find(any(Bson.class))).thenReturn(result);
+ when(result.projection(any())).thenReturn(result);
+ when(result.first()).thenReturn(collectionMetadata);
+
+ Assertions.assertSame(
+ collectionMetadata,
+ MongodbUtils.readCollectionMetadata(
+ mongoClient, TableId.parse("inventory.products")));
+
+ org.mockito.ArgumentCaptor<Bson> projectionCaptor =
+ org.mockito.ArgumentCaptor.forClass(Bson.class);
+ verify(result).projection(projectionCaptor.capture());
+ BsonDocument projection =
+ projectionCaptor
+ .getValue()
+ .toBsonDocument(
+ BsonDocument.class,
MongoClientSettings.getDefaultCodecRegistry());
+ Assertions.assertEquals(new BsonInt32(1),
projection.get(SHARD_KEY_FIELD));
+ }
+
@Test
public void testBuildConnectionStringFromHostsKeepsLegacyFormat() {
String connectionString =
