JNSimba commented on code in PR #167:
URL:
https://github.com/apache/doris-flink-connector/pull/167#discussion_r1281853617
##########
flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/JsonDebeziumSchemaSerializer.java:
##########
@@ -113,6 +138,67 @@ public byte[] serialize(String record) throws IOException {
return
objectMapper.writeValueAsString(valueMap).getBytes(StandardCharsets.UTF_8);
}
+ public boolean schemaChangeV2(JsonNode recordRoot) {
+ boolean status = false;
+ try {
+ if (!StringUtils.isNullOrWhitespaceOnly(sourceTableName) &&
!checkTable(recordRoot)) {
+ return false;
+ }
+ List<String> ddlSqlList = extractDDLList(recordRoot);
+ if (CollectionUtils.isEmpty(ddlSqlList)) {
+ LOG.info("ddl can not do schema change:{}", recordRoot);
+ return false;
+ }
+
+ List<DDLSchema> ddlSchemas = SchemaChangeHelper.getDdlSchemas();
+ for (int i = 0; i < ddlSqlList.size(); i++) {
+ DDLSchema ddlSchema = ddlSchemas.get(i);
+ String ddlSql = ddlSqlList.get(i);
+ boolean doSchemaChange = checkSchemaChange(ddlSchema);
+ status = doSchemaChange && execSchemaChange(ddlSql);
+ LOG.info("schema change status:{}", status);
+ }
+ } catch (Exception ex) {
+ LOG.warn("schema change error :", ex);
+ }
+ return status;
+ }
+
+ private boolean checkSchemaChange(DDLSchema ddlSchema) throws IOException,
IllegalArgumentException {
+ String requestUrl = String.format(CHECK_SCHEMA_CHANGE_API,
RestService.randomEndpoint(dorisOptions.getFenodes(), LOG), database, table);
+ Map<String,Object> param = buildRequestParam(ddlSchema);
+ HttpGetWithEntity httpGet = new HttpGetWithEntity(requestUrl);
+ httpGet.setHeader(HttpHeaders.AUTHORIZATION, authHeader());
+ httpGet.setEntity(new
StringEntity(objectMapper.writeValueAsString(param)));
+ boolean success = handleResponse(httpGet);
+ if (!success) {
+ LOG.warn("schema change can not do table {}.{}",database,table);
+ }
+ return success;
+ }
+
+ private List<String> extractDDLList(JsonNode record) throws
JsonProcessingException {
Review Comment:
Need to add unit tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
