This is an automated email from the ASF dual-hosted git repository.
fanjia 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 1870653393 [Improve][CDC] Filter heartbeat event (#8569)
1870653393 is described below
commit 187065339318e52d036efb1621d675474d9785f0
Author: hailin0 <[email protected]>
AuthorDate: Thu Feb 27 16:37:56 2025 +0800
[Improve][CDC] Filter heartbeat event (#8569)
---
.../row/DebeziumJsonDeserializeSchema.java | 10 ++-
.../row/DebeziumJsonDeserializeSchemaTest.java | 91 ++++++++++++++++++++++
2 files changed, 99 insertions(+), 2 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/DebeziumJsonDeserializeSchema.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/DebeziumJsonDeserializeSchema.java
index 47b1413bd7..9e5215203f 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/DebeziumJsonDeserializeSchema.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/DebeziumJsonDeserializeSchema.java
@@ -34,6 +34,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static
org.apache.seatunnel.connectors.cdc.base.utils.SourceRecordUtils.isHeartbeatRecord;
+
@Slf4j
public class DebeziumJsonDeserializeSchema
extends AbstractDebeziumDeserializationSchema<SeaTunnelRow> {
@@ -60,9 +62,13 @@ public class DebeziumJsonDeserializeSchema
@Override
public void deserialize(SourceRecord record, Collector<SeaTunnelRow> out)
throws Exception {
super.deserialize(record, out);
+ if (!isHeartbeatRecord(record)) {
+ SeaTunnelRow row = deserializationSchema.deserialize(record);
+ out.collect(row);
+ return;
+ }
- SeaTunnelRow row = deserializationSchema.deserialize(record);
- out.collect(row);
+ log.debug("Unsupported record {}, just skip.", record);
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/debezium/row/DebeziumJsonDeserializeSchemaTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/debezium/row/DebeziumJsonDeserializeSchemaTest.java
new file mode 100644
index 0000000000..482da1eaac
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/debezium/row/DebeziumJsonDeserializeSchemaTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.seatunnel.connectors.cdc.debezium.row;
+
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class DebeziumJsonDeserializeSchemaTest {
+ @Test
+ void deserializeNonHeartbeatRecord() throws Exception {
+ Map<String, String> debeziumConfig = Collections.EMPTY_MAP;
+ DebeziumJsonDeserializeSchema schema = new
DebeziumJsonDeserializeSchema(debeziumConfig);
+
+ // Create a schema for the record
+ SchemaBuilder schemaBuilder =
+ SchemaBuilder.struct()
+ .name("test")
+ .field("field",
SchemaBuilder.string().optional().build());
+ Struct struct = new Struct(schemaBuilder.build()).put("field",
"value");
+ SourceRecord record =
+ new SourceRecord(
+ null,
+ null,
+ "test",
+ schemaBuilder.build(),
+ struct,
+ schemaBuilder.build(),
+ struct);
+
+ Collector<SeaTunnelRow> collector = mock(Collector.class);
+ schema.deserialize(record, collector);
+
+ verify(collector, times(1)).collect(any(SeaTunnelRow.class));
+ }
+
+ @Test
+ void skipHeartbeatRecord() throws Exception {
+ Map<String, String> debeziumConfig = Collections.EMPTY_MAP;
+ DebeziumJsonDeserializeSchema schema = new
DebeziumJsonDeserializeSchema(debeziumConfig);
+
+ // Create a schema for the record
+ SchemaBuilder schemaBuilder =
+ SchemaBuilder.struct()
+ .name("io.debezium.connector.common.Heartbeat")
+ .field("field",
SchemaBuilder.string().optional().build());
+ Struct struct = new Struct(schemaBuilder.build()).put("field",
"value");
+ SourceRecord record =
+ new SourceRecord(
+ null,
+ null,
+ "test",
+ schemaBuilder.build(),
+ struct,
+ schemaBuilder.build(),
+ struct);
+
+ Collector<SeaTunnelRow> collector = mock(Collector.class);
+ schema.deserialize(record, collector);
+
+ verify(collector, times(0)).collect(any(SeaTunnelRow.class));
+ }
+}
