[fix]fix SchemaChangeManager cannot do schema change (apache#241)

JNSimba · Nov 23, 2023 · 7dd8512 · 7dd8512
1 parent 2b523f4
commit 7dd8512
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 7 deletions.
diff --git a/...doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeManager.java b/...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/...main/java/org/apache/doris/flink/sink/writer/serializer/JsonDebeziumSchemaSerializer.java b/...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);