From f4f8b14f624b2e90de543c8fcd1c9480cd82d4fc Mon Sep 17 00:00:00 2001
From: Asura7969 <1402357969@qq.com>
Date: Mon, 8 Jul 2024 12:49:43 +0800
Subject: [PATCH] [Fix][Connector-V2][CDC]
 SeaTunnelRowDebeziumDeserializationConverters NPE (#7119)

---
 .../debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java   | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java
index 1fd706290185..d09e7b77b5c1 100644
--- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java
+++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java
@@ -184,7 +184,7 @@ private void deserializeDataChangeRecord(SourceRecord record, Collector<SeaTunne
             delete.setRowKind(RowKind.DELETE);
             delete.setTableId(tableId);
             collector.collect(delete);
-        } else {
+        } else if (operation == Envelope.Operation.UPDATE) {
             SeaTunnelRow before = extractBeforeRow(converters, record, messageStruct, valueSchema);
             before.setRowKind(RowKind.UPDATE_BEFORE);
             before.setTableId(tableId);
@@ -194,6 +194,8 @@ private void deserializeDataChangeRecord(SourceRecord record, Collector<SeaTunne
             after.setRowKind(RowKind.UPDATE_AFTER);
             after.setTableId(tableId);
             collector.collect(after);
+        } else {
+            log.warn("Received {} operation, skip", operation);
         }
     }