JNSimba commented on code in PR #422:
URL:
https://github.com/apache/doris-flink-connector/pull/422#discussion_r1678878401
##########
flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChange.java:
##########
@@ -120,6 +137,57 @@ protected JsonNode extractHistoryRecord(JsonNode record)
throws JsonProcessingEx
return record;
}
+ /** Parse event type. */
+ protected EventType extractEventType(JsonNode record) throws
JsonProcessingException {
+ JsonNode tableChange = extractTableChange(record);
+ if (tableChange == null || tableChange.get("type") == null) {
+ return null;
+ }
+ String type = tableChange.get("type").asText();
+ if (EventType.ALTER.toString().equalsIgnoreCase(type)) {
+ return EventType.ALTER;
+ } else if (EventType.CREATE.toString().equalsIgnoreCase(type)) {
+ return EventType.CREATE;
+ }
+ return null;
+ }
+
+ protected JsonNode extractTableChange(JsonNode record) throws
JsonProcessingException {
+ JsonNode historyRecord = extractHistoryRecord(record);
+ JsonNode tableChanges = historyRecord.get("tableChanges");
+ if (Objects.nonNull(tableChanges)) {
+ return tableChanges.get(0);
+ }
+ LOG.warn("Failed to extract tableChanges. record={}", record);
+ return null;
+ }
+
+ protected boolean executeAlterDDLs(
+ List<String> ddlSqlList,
+ JsonNode recordRoot,
+ Tuple2<String, String> dorisTableTuple,
+ boolean status)
+ throws IOException, IllegalArgumentException {
+ if (CollectionUtils.isEmpty(ddlSqlList)) {
+ LOG.info("The recordRoot cannot extract ddl sql. recordRoot={}",
recordRoot);
+ return false;
+ }
+
+ for (String ddlSql : ddlSqlList) {
+ status = schemaChangeManager.execute(ddlSql, dorisTableTuple.f0);
+ LOG.info("schema change status:{}, ddl: {}", status, ddlSql);
+ }
+ return status;
+ }
+
+ protected void extractSourceConnector(JsonNode record) {
+ if (Objects.isNull(sourceConnector)) {
+ sourceConnector =
Review Comment:
Will sourceConnector be null?
--
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
