From 0aa121d19f22f63abd73134e44e1e54f22c45ea1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E6=B5=B7=E7=9B=90=E5=84=BF?= <leeyating@hust.edu.cn>
Date: Fri, 21 Apr 2023 19:38:52 +0800
Subject: [PATCH] Fix_NullPointerException

---
 .../java/org/gauss/util/DDLProcessor.java     | 28 ++++++++------
 .../java/org/gauss/util/ddl/DDLProcessor.java | 37 ++++++++++++-------
 2 files changed, 39 insertions(+), 26 deletions(-)

diff --git a/src/main/java/org/gauss/util/DDLProcessor.java b/src/main/java/org/gauss/util/DDLProcessor.java
index 8ad906d..d39fb96 100644
--- a/src/main/java/org/gauss/util/DDLProcessor.java
+++ b/src/main/java/org/gauss/util/DDLProcessor.java
@@ -32,20 +32,24 @@ public class DDLProcessor extends Processor {
         try {
             value = topicMapper.readValue(record.value(), DDLValueStruct.class);
         } catch (Exception e) {
-            e.printStackTrace();
+            LOGGER.info(e.getMessage());
         }
 
-        SourceStruct source = value.getPayload().getSource();
-        String snapshot = source.getSnapshot();
-        if (snapshot.equals("true") || snapshot.equals("last")) {
-            return;
+        if (value != null && value.getPayload() != null && value.getPayload().getSource() != null) {
+            SourceStruct source = value.getPayload().getSource();
+            String snapshot = source.getSnapshot();
+            if ("true".equals(snapshot) || "last".equals(snapshot)) {
+                return;
+            }
+
+            String ddl = ddlSqlParser.parse(value);
+            LOGGER.info("execute DDL SQL: {}", ddl);
+            // We don't do heavy work on DDL and just pass the origin DDL SQL to destination
+            // database. We assume some compatibility plugins in destination database may
+            // process these DDL SQL.
+            executor.executeDDL(ddl);
+        } else {
+            LOGGER.warn("Invalid DDLValueStruct received: {}", record.value());
         }
-
-        String ddl = ddlSqlParser.parse(value);
-        LOGGER.info("execute DDL SQl: {}", ddl);
-        // We don't do heavy work on DDL and just pass the origin DDL SQL to destination
-        // database. We assume some compatibility plugins in destination database may
-        // process these DDL SQL.
-        executor.executeDDL(ddl);
     }
 }
diff --git a/src/main/java/org/gauss/util/ddl/DDLProcessor.java b/src/main/java/org/gauss/util/ddl/DDLProcessor.java
index e109070..bcabb5f 100644
--- a/src/main/java/org/gauss/util/ddl/DDLProcessor.java
+++ b/src/main/java/org/gauss/util/ddl/DDLProcessor.java
@@ -34,31 +34,40 @@ public class DDLProcessor extends Processor {
         try {
             value = topicMapper.readValue(record.value(), DDLValueStruct.class);
         } catch (Exception e) {
-            e.printStackTrace();
+            LOGGER.info(e.getMessage());
         }
-        SourceStruct source = value.getPayload().getSource();
-        String snapshot = source.getSnapshot();
-        if (snapshot.equals("true") || snapshot.equals("last")) {
-            return;
+
+        if (value != null && value.getPayload() != null && value.getPayload().getSource() != null) {
+            SourceStruct source = value.getPayload().getSource();
+            String snapshot = source.getSnapshot();
+            if ("true".equals(snapshot) || "last".equals(snapshot)) {
+                return;
+            }
+
+            String ddl = value.getPayload().getDdl();
+            LOGGER.info("get DDL SQL: {}", ddl);
+        } else {
+            LOGGER.warn("Invalid DDLValueStruct received: {}", record.value());
         }
-        String ddl = value.getPayload().getDdl();
-        LOGGER.info("get DDL SQL: {}", ddl);
 
         // sometimes one oracle ddl sql may convert to many opengauss ddl;
         List<DDLConvert> dDlConverts = DDLConvertHandler.getDDlConvert(value.getPayload());
         for (DDLConvert dDlConvert : dDlConverts) {
             List<String> openGaussDDL = dDlConvert.convertToOpenGaussDDL(value);
-            // when get ddl sql ,if sql contains rename table ,drop table, drop column , must compare current dml scn and ddl scn
-            if (null != openGaussDDL) {
+            // when get ddl sql, if sql contains rename table, drop table,
+            // drop column, must compare current dml scn and ddl scn
+            if (openGaussDDL != null) {
                 if (dDlConvert.needCacheSql()) {
-                    // some time the field commit_scn will be null ,so we use scn field, if scn and commit_scn both have value, we use Math.min(scn,commit_scn) as value
+                    // some time the field commitScn will be null, so
+                    // we use scn field, if scn and commitScn both have
+                    // value, we use Math.min(scn,commitScn) as value
                     long currentScn;
-                    String commit_scn = value.getPayload().getSource().getCommit_scn();
+                    String commitScn = value.getPayload().getSource().getCommit_scn();
                     Long scn = value.getPayload().getSource().getScn();
-                    if (commit_scn != null && scn != null) {
-                        currentScn = Math.min(scn, Long.parseLong(commit_scn));
+                    if (commitScn != null && scn != null) {
+                        currentScn = Math.min(scn, Long.parseLong(commitScn));
                     } else {
-                        currentScn = commit_scn != null ? Long.parseLong(commit_scn) : scn;
+                        currentScn = commitScn != null ? Long.parseLong(commitScn) : scn;
                     }
                     LOGGER.info("add ddl sql: {} to cache,ddl_scn: {}", openGaussDDL, currentScn);
                     ddlCacheController.addDdls(currentScn, openGaussDDL);
-- 
Gitee