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 ebae5c10c3 [Fix][Connector-V2] Fix MongoDB CDC missing operation type
handling (#11208)
ebae5c10c3 is described below
commit ebae5c10c3c2abacd8a8d872c6e5d590d353c322
Author: Jast <[email protected]>
AuthorDate: Tue Jul 21 22:44:35 2026 +0800
[Fix][Connector-V2] Fix MongoDB CDC missing operation type handling (#11208)
---
.../MongoDBConnectorDeserializationSchema.java | 34 +---------
.../source/fetch/MongodbFetchTaskContext.java | 34 +---------
.../cdc/mongodb/utils/MongodbRecordUtils.java | 28 +++++++++
.../MongoDBConnectorDeserializationSchemaTest.java | 73 ++++++++++++++++++++++
4 files changed, 106 insertions(+), 63 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/sender/MongoDBConnectorDeserializationSchema.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/sender/MongoDBConnectorDeserializationSchema.java
index d762c36161..1ff9410dfd 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/sender/MongoDBConnectorDeserializationSchema.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/sender/MongoDBConnectorDeserializationSchema.java
@@ -75,8 +75,8 @@ import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.Mongo
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.ENCODE_VALUE_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.FULL_DOCUMENT;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.NS_FIELD;
-import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.OPERATION_TYPE;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils.extractBsonDocument;
+import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils.getOperationType;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils.isHeartbeatEvent;
import static
org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkNotNull;
@@ -110,7 +110,7 @@ public class MongoDBConnectorDeserializationSchema
Struct value = (Struct) record.value();
Schema valueSchema = record.valueSchema();
- OperationType op = operationTypeFor(record);
+ OperationType op = getOperationType(record);
BsonDocument documentKey =
checkNotNull(
Objects.requireNonNull(
@@ -184,36 +184,6 @@ public class MongoDBConnectorDeserializationSchema
return tables;
}
- private @Nonnull OperationType operationTypeFor(@Nonnull SourceRecord
record) {
- Struct value = (Struct) record.value();
- Schema valueSchema = record.valueSchema();
- if (valueSchema == null || valueSchema.field(OPERATION_TYPE) == null) {
- throw new MongodbConnectorException(
- ILLEGAL_ARGUMENT,
- String.format(
- "MongoDB CDC record has no %s field. Topic: %s,
partition: %s,"
- + " offset: %s",
- OPERATION_TYPE,
- record.topic(),
- record.sourcePartition(),
- record.sourceOffset()));
- }
-
- String operationType = value.getString(OPERATION_TYPE);
- if (operationType == null) {
- throw new MongodbConnectorException(
- ILLEGAL_ARGUMENT,
- String.format(
- "MongoDB CDC record has null %s field. Topic: %s,
partition: %s,"
- + " offset: %s",
- OPERATION_TYPE,
- record.topic(),
- record.sourcePartition(),
- record.sourceOffset()));
- }
- return OperationType.fromString(operationType);
- }
-
// TODO:The dynamic schema will be completed based on this method later.
private void emit(
SourceRecord inRecord,
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbFetchTaskContext.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbFetchTaskContext.java
index 91fc9ec769..e018036389 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbFetchTaskContext.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbFetchTaskContext.java
@@ -71,6 +71,7 @@ import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.BsonUt
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils.buildSourceRecord;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils.extractBsonDocument;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils.getDocumentKey;
+import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils.getOperationType;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils.getResumeToken;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils.isHeartbeatEvent;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbUtils.createMongoClient;
@@ -212,9 +213,8 @@ public class MongodbFetchTaskContext implements
FetchTask.Context {
Struct value = (Struct) changeRecord.value();
if (value != null) {
- String operationType = getOperationType(changeRecord, value);
-
- switch (OperationType.fromString(operationType)) {
+ OperationType operationType = getOperationType(changeRecord);
+ switch (operationType) {
case INSERT:
outputBuffer.put(key, changeRecord);
break;
@@ -249,34 +249,6 @@ public class MongodbFetchTaskContext implements
FetchTask.Context {
}
}
- private String getOperationType(SourceRecord record, Struct value) {
- if (record.valueSchema() == null ||
record.valueSchema().field(OPERATION_TYPE) == null) {
- throw new MongodbConnectorException(
- ILLEGAL_ARGUMENT,
- String.format(
- "MongoDB CDC record has no %s field. Topic: %s,
partition: %s,"
- + " offset: %s",
- OPERATION_TYPE,
- record.topic(),
- record.sourcePartition(),
- record.sourceOffset()));
- }
-
- String operationType = value.getString(OPERATION_TYPE);
- if (operationType == null) {
- throw new MongodbConnectorException(
- ILLEGAL_ARGUMENT,
- String.format(
- "MongoDB CDC record has null %s field. Topic: %s,
partition: %s,"
- + " offset: %s",
- OPERATION_TYPE,
- record.topic(),
- record.sourcePartition(),
- record.sourceOffset()));
- }
- return operationType;
- }
-
@Override
public List<SourceRecord> formatMessageTimestamp(
@Nonnull Collection<SourceRecord> snapshotRecords) {
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbRecordUtils.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbRecordUtils.java
index 9b378df424..d6a2a406e4 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbRecordUtils.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbRecordUtils.java
@@ -19,6 +19,8 @@ package
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils;
import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.exception.MongodbConnectorException;
+
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaAndValue;
import org.apache.kafka.connect.data.Struct;
@@ -29,6 +31,7 @@ import org.bson.BsonTimestamp;
import org.bson.BsonValue;
import org.bson.json.JsonWriterSettings;
+import com.mongodb.client.model.changestream.OperationType;
import com.mongodb.kafka.connect.source.json.formatter.DefaultJson;
import com.mongodb.kafka.connect.source.schema.AvroSchemaDefaults;
import com.mongodb.kafka.connect.source.schema.BsonValueToSchemaAndValue;
@@ -43,6 +46,7 @@ import java.util.Map;
import static com.mongodb.kafka.connect.source.schema.AvroSchema.fromJson;
import static io.debezium.connector.AbstractSourceInfo.TABLE_NAME_KEY;
+import static
org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT;
import static
org.apache.seatunnel.connectors.cdc.base.source.split.wartermark.WatermarkEvent.isWatermarkEvent;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.COLL_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.COPY_KEY_FIELD;
@@ -51,6 +55,7 @@ import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.Mongo
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.HEARTBEAT_KEY_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.ID_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.NS_FIELD;
+import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.OPERATION_TYPE;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.OUTPUT_SCHEMA;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.SOURCE_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbUtils.buildConnectionNamespacePrefix;
@@ -71,6 +76,29 @@ public class MongodbRecordUtils {
return !isWatermarkEvent(sourceRecord) &&
!isHeartbeatEvent(sourceRecord);
}
+ public static @Nonnull OperationType getOperationType(@Nonnull
SourceRecord sourceRecord) {
+ Struct value = (Struct) sourceRecord.value();
+ if (value == null) {
+ throw new MongodbConnectorException(
+ ILLEGAL_ARGUMENT,
+ "MongoDB CDC data change record has no value: " +
sourceRecord);
+ }
+
+ String operationType = null;
+ if (value.schema().field(OPERATION_TYPE) != null) {
+ operationType = value.getString(OPERATION_TYPE);
+ }
+ if (StringUtils.isEmpty(operationType)) {
+ if (isSnapshotRecord(sourceRecord)) {
+ return OperationType.INSERT;
+ }
+ throw new MongodbConnectorException(
+ ILLEGAL_ARGUMENT,
+ "MongoDB CDC data change record has no operationType
field: " + sourceRecord);
+ }
+ return OperationType.fromString(operationType);
+ }
+
public static BsonDocument getResumeToken(SourceRecord sourceRecord) {
return BsonDocument.parse(getOffsetValue(sourceRecord, ID_FIELD));
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/sender/MongoDBConnectorDeserializationSchemaTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/sender/MongoDBConnectorDeserializationSchemaTest.java
index 8f5be4dcd6..12f8672373 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/sender/MongoDBConnectorDeserializationSchemaTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/sender/MongoDBConnectorDeserializationSchemaTest.java
@@ -25,6 +25,7 @@ import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.DecimalType;
import org.apache.seatunnel.api.table.type.LocalTimeType;
+import org.apache.seatunnel.api.table.type.RowKind;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.exception.MongodbConnectorException;
@@ -174,6 +175,22 @@ public class MongoDBConnectorDeserializationSchemaTest {
Assertions.assertEquals("inventory.products", tableId);
}
+ @Test
+ public void deserializeSnapshotRecordWithoutOperationType() throws
Exception {
+ MongoDBConnectorDeserializationSchema schema =
+ new
MongoDBConnectorDeserializationSchema(Collections.singletonList(catalogTable));
+ SourceRecord sourceRecord = createSnapshotRecordWithoutOperationType();
+ List<SeaTunnelRow> rows = new ArrayList<>();
+
+ schema.deserialize(sourceRecord, new ListCollector(rows));
+
+ Assertions.assertEquals(1, rows.size());
+ SeaTunnelRow row = rows.get(0);
+ Assertions.assertEquals(RowKind.INSERT, row.getRowKind());
+ Assertions.assertEquals("database.table", row.getTableId());
+ Assertions.assertEquals("snapshot-value", row.getField(4));
+ }
+
@Test
public void skipHeartbeatRecordWithoutOperationType() throws Exception {
MongoDBConnectorDeserializationSchema schema =
@@ -303,6 +320,62 @@ public class MongoDBConnectorDeserializationSchemaTest {
.orElseThrow(() -> new RuntimeException("not found field"));
}
+ private SourceRecord createSnapshotRecordWithoutOperationType() {
+ Schema keySchema = SchemaBuilder.struct().field(ID_FIELD,
Schema.STRING_SCHEMA).build();
+ Struct key = new Struct(keySchema).put(ID_FIELD, "12");
+ Schema sourceSchema =
+ SchemaBuilder.struct()
+ .name("source")
+ .field(TS_MS_FIELD, Schema.INT64_SCHEMA)
+ .field("table", Schema.OPTIONAL_STRING_SCHEMA)
+ .field(DB_FIELD, Schema.OPTIONAL_STRING_SCHEMA)
+ .field(SNAPSHOT_FIELD, Schema.OPTIONAL_STRING_SCHEMA)
+ .build();
+ Schema nsSchema =
+ SchemaBuilder.struct()
+ .name("ns")
+ .field(DB_FIELD, Schema.STRING_SCHEMA)
+ .field(COLL_FIELD, Schema.OPTIONAL_STRING_SCHEMA)
+ .build();
+ Schema valueSchema =
+ SchemaBuilder.struct()
+ .name("ChangeStreamWithoutOperationType")
+ .field(ID_FIELD, Schema.STRING_SCHEMA)
+ .field(FULL_DOCUMENT, Schema.OPTIONAL_STRING_SCHEMA)
+ .field(SOURCE_FIELD, sourceSchema)
+ .field(TS_MS_FIELD, Schema.OPTIONAL_INT64_SCHEMA)
+ .field(NS_FIELD, nsSchema)
+ .field(DOCUMENT_KEY, Schema.OPTIONAL_STRING_SCHEMA)
+ .build();
+
+ Struct source =
+ new Struct(sourceSchema)
+ .put(TS_MS_FIELD, 0L)
+ .put("table", "table")
+ .put(DB_FIELD, "database")
+ .put(SNAPSHOT_FIELD, SNAPSHOT_TRUE);
+ Struct ns = new Struct(nsSchema).put(DB_FIELD,
"database").put(COLL_FIELD, "table");
+ BsonDocument documentKey = new BsonDocument(ID_FIELD, new
BsonString("12"));
+ BsonDocument fullDocument = new BsonDocument("string", new
BsonString("snapshot-value"));
+ Struct value =
+ new Struct(valueSchema)
+ .put(ID_FIELD, documentKey.toJson())
+ .put(FULL_DOCUMENT, fullDocument.toJson())
+ .put(SOURCE_FIELD, source)
+ .put(TS_MS_FIELD, System.currentTimeMillis())
+ .put(NS_FIELD, ns)
+ .put(DOCUMENT_KEY, documentKey.toJson());
+
+ return new SourceRecord(
+ MongodbRecordUtils.createPartitionMap("localhost:27017",
"database", "table"),
+ createSourceOffsetMap(documentKey, true),
+ "database.table",
+ keySchema,
+ key,
+ valueSchema,
+ value);
+ }
+
private SourceRecord createRecordWithoutOperationType(boolean heartbeat) {
Schema valueSchema = SchemaBuilder.struct().field(TS_MS_FIELD,
Schema.INT64_SCHEMA).build();
Struct value = new Struct(valueSchema).put(TS_MS_FIELD,
System.currentTimeMillis());
