This is an automated email from the ASF dual-hosted git repository.
zhangshenghang 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 f570a86597 [Fix][Connector-V2] Skip MongoDB CDC heartbeat records
(#11065)
f570a86597 is described below
commit f570a8659733e3bc44d99a4d6ce80ea50b3fc73f
Author: Daniel Carter <[email protected]>
AuthorDate: Fri Jul 3 08:10:32 2026 +0800
[Fix][Connector-V2] Skip MongoDB CDC heartbeat records (#11065)
Co-authored-by: Shenghang <[email protected]>
Co-authored-by: DanielCarter-stack
<[email protected]>
Co-authored-by: DanielCarter-stack
<[email protected]>
---
.../MongoDBConnectorDeserializationSchema.java | 33 +++++++++-
.../source/fetch/MongodbFetchTaskContext.java | 30 ++++++++-
.../MongoDBConnectorDeserializationSchemaTest.java | 72 ++++++++++++++++++++++
3 files changed, 133 insertions(+), 2 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 6b14d082c9..d762c36161 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,7 +75,9 @@ 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.isHeartbeatEvent;
import static
org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkNotNull;
@Slf4j
@@ -101,6 +103,10 @@ public class MongoDBConnectorDeserializationSchema
throws Exception {
super.deserialize(record, out);
+ if (isHeartbeatEvent(record)) {
+ return;
+ }
+
Struct value = (Struct) record.value();
Schema valueSchema = record.valueSchema();
@@ -180,7 +186,32 @@ public class MongoDBConnectorDeserializationSchema
private @Nonnull OperationType operationTypeFor(@Nonnull SourceRecord
record) {
Struct value = (Struct) record.value();
- return OperationType.fromString(value.getString("operationType"));
+ 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.
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 a8d5538181..91fc9ec769 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
@@ -212,7 +212,7 @@ public class MongodbFetchTaskContext implements
FetchTask.Context {
Struct value = (Struct) changeRecord.value();
if (value != null) {
- String operationType = value.getString(OPERATION_TYPE);
+ String operationType = getOperationType(changeRecord, value);
switch (OperationType.fromString(operationType)) {
case INSERT:
@@ -249,6 +249,34 @@ 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/test/java/mongodb/sender/MongoDBConnectorDeserializationSchemaTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/sender/MongoDBConnectorDeserializationSchemaTest.java
index c64952c95c..8f5be4dcd6 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
@@ -17,6 +17,7 @@
package mongodb.sender;
+import org.apache.seatunnel.api.source.Collector;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
@@ -25,10 +26,14 @@ 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.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.exception.MongodbConnectorException;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.sender.MongoDBConnectorDeserializationSchema;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.bson.BsonDateTime;
@@ -49,7 +54,10 @@ import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.IntStream;
@@ -58,6 +66,7 @@ import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.Mongo
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.DB_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.DOCUMENT_KEY;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.FULL_DOCUMENT;
+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;
@@ -165,6 +174,32 @@ public class MongoDBConnectorDeserializationSchemaTest {
Assertions.assertEquals("inventory.products", tableId);
}
+ @Test
+ public void skipHeartbeatRecordWithoutOperationType() throws Exception {
+ MongoDBConnectorDeserializationSchema schema =
+ new
MongoDBConnectorDeserializationSchema(Collections.singletonList(catalogTable));
+ SourceRecord heartbeatRecord = createRecordWithoutOperationType(true);
+ List<SeaTunnelRow> rows = new ArrayList<>();
+
+ schema.deserialize(heartbeatRecord, new ListCollector(rows));
+
+ Assertions.assertTrue(rows.isEmpty());
+ }
+
+ @Test
+ public void throwReadableExceptionWhenOperationTypeMissing() {
+ MongoDBConnectorDeserializationSchema schema =
+ new
MongoDBConnectorDeserializationSchema(Collections.singletonList(catalogTable));
+ SourceRecord record = createRecordWithoutOperationType(false);
+
+ MongodbConnectorException exception =
+ Assertions.assertThrows(
+ MongodbConnectorException.class,
+ () -> schema.deserialize(record, new ListCollector(new
ArrayList<>())));
+
+ Assertions.assertTrue(exception.getMessage().contains(OPERATION_TYPE));
+ }
+
@Test
public void testBsonConvert() {
MongoDBConnectorDeserializationSchema schema =
@@ -267,4 +302,41 @@ public class MongoDBConnectorDeserializationSchemaTest {
.findFirst()
.orElseThrow(() -> new RuntimeException("not found field"));
}
+
+ 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());
+ Map<String, String> sourceOffset = new HashMap<>();
+ if (heartbeat) {
+ sourceOffset.put(HEARTBEAT_KEY_FIELD, "true");
+ }
+
+ return new SourceRecord(
+ Collections.singletonMap(
+ NS_FIELD,
"mongodb://localhost:27017/__mongodb_heartbeats"),
+ sourceOffset,
+ "__mongodb_heartbeats",
+ null,
+ null,
+ valueSchema,
+ value);
+ }
+
+ private static class ListCollector implements Collector<SeaTunnelRow> {
+ private final List<SeaTunnelRow> rows;
+
+ private ListCollector(List<SeaTunnelRow> rows) {
+ this.rows = rows;
+ }
+
+ @Override
+ public void collect(SeaTunnelRow record) {
+ rows.add(record);
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return this;
+ }
+ }
}
