This is an automated email from the ASF dual-hosted git repository.
diwu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-flink-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 7dd8512 [fix]fix SchemaChangeManager cannot do schema change (#241)
7dd8512 is described below
commit 7dd8512ea9ab870147277cd95833068172e85c19
Author: wudongliang <46414265+donglian...@users.noreply.github.com>
AuthorDate: Thu Nov 23 16:20:49 2023 +0800
[fix]fix SchemaChangeManager cannot do schema change (#241)
---
.../apache/doris/flink/sink/schema/SchemaChangeManager.java | 11 ++++++-----
.../sink/writer/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 aff30c9..d9c3345 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 class SchemaChangeManager implements Serializable {
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 class SchemaChangeManager implements Serializable {
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 class SchemaChangeManager implements Serializable {
/**
* 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 ee4cad3..50ae7fc 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 class JsonDebeziumSchemaSerializer implements
DorisRecordSerializer<Strin
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 class JsonDebeziumSchemaSerializer implements
DorisRecordSerializer<Strin
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);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
