From 7dd8512ea9ab870147277cd95833068172e85c19 Mon Sep 17 00:00:00 2001
From: wudongliang <46414265+DongLiang-0@users.noreply.github.com>
Date: Thu, 23 Nov 2023 16:20:49 +0800
Subject: [PATCH] [fix]fix SchemaChangeManager cannot do schema change (#241)

---
 .../doris/flink/sink/schema/SchemaChangeManager.java  | 11 ++++++-----
 .../serializer/JsonDebeziumSchemaSerializer.java      |  4 ++--
 2 files changed, 8 insertions(+), 7 deletions(-)

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeManager.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeManager.java
index aff30c9aa..d9c3345d5 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeManager.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeManager.java
@@ -49,7 +49,7 @@ public class SchemaChangeManager implements Serializable {
     private static final long serialVersionUID = 1L;
     private static final Logger LOG = LoggerFactory.getLogger(SchemaChangeManager.class);
     private static final String CHECK_SCHEMA_CHANGE_API = "http://%s/api/enable_light_schema_change/%s/%s";
-    private static final String SCHEMA_CHANGE_API = "http://%s/api/query/default_cluster/";
+    private static final String SCHEMA_CHANGE_API = "http://%s/api/query/default_cluster/%s";
     private ObjectMapper objectMapper = new ObjectMapper();
     private DorisOptions dorisOptions;
 
@@ -59,7 +59,7 @@ public SchemaChangeManager(DorisOptions dorisOptions) {
 
     public boolean createTable(TableSchema table) throws IOException, IllegalArgumentException {
         String createTableDDL = DorisSystem.buildCreateTableDDL(table);
-        return execute(createTableDDL);
+        return execute(createTableDDL, table.getDatabase());
     }
 
     public boolean addColumn(String database, String table, FieldSchema field) throws IOException, IllegalArgumentException {
@@ -82,7 +82,7 @@ public boolean renameColumn(String database, String table, String oldColumnName,
 
     public boolean schemaChange(String database, String table, Map<String, Object> params, String sql) throws IOException, IllegalArgumentException {
         if(checkSchemaChange(database, table, params)){
-            return execute(sql);
+            return execute(sql, database);
         }
         return false;
     }
@@ -116,13 +116,14 @@ public boolean checkSchemaChange(String database, String table, Map<String, Obje
     /**
      * execute sql in doris
      */
-    public boolean execute(String ddl) throws IOException, IllegalArgumentException {
+    public boolean execute(String ddl, String database) throws IOException, IllegalArgumentException {
         if(StringUtils.isNullOrWhitespaceOnly(ddl)){
             return false;
         }
         Map<String, String> param = new HashMap<>();
         param.put("stmt", ddl);
-        String requestUrl = String.format(SCHEMA_CHANGE_API, RestService.randomEndpoint(dorisOptions.getFenodes(), LOG));
+        String requestUrl = String.format(SCHEMA_CHANGE_API,
+                RestService.randomEndpoint(dorisOptions.getFenodes(), LOG), database);
         HttpPost httpPost = new HttpPost(requestUrl);
         httpPost.setHeader(HttpHeaders.AUTHORIZATION, authHeader());
         httpPost.setHeader(HttpHeaders.CONTENT_TYPE, "application/json");
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/JsonDebeziumSchemaSerializer.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/JsonDebeziumSchemaSerializer.java
index ee4cad3c7..50ae7fcc9 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/JsonDebeziumSchemaSerializer.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/JsonDebeziumSchemaSerializer.java
@@ -198,7 +198,7 @@ public boolean schemaChangeV2(JsonNode recordRoot) {
                 DDLSchema ddlSchema = ddlSchemas.get(i);
                 String ddlSql = ddlSqlList.get(i);
                 boolean doSchemaChange = checkSchemaChange(ddlSchema);
-                status = doSchemaChange && schemaChangeManager.execute(ddlSql);
+                status = doSchemaChange && schemaChangeManager.execute(ddlSql, database);
                 LOG.info("schema change status:{}, ddl:{}", status, ddlSql);
             }
         } catch (Exception ex) {
@@ -268,7 +268,7 @@ public boolean schemaChange(JsonNode recordRoot) {
                 return false;
             }
             boolean doSchemaChange = checkSchemaChange(ddl);
-            status = doSchemaChange && schemaChangeManager.execute(ddl);
+            status = doSchemaChange && schemaChangeManager.execute(ddl, database);
             LOG.info("schema change status:{}", status);
         } catch (Exception ex) {
             LOG.warn("schema change error :", ex);