Copilot commented on code in PR #61182:
URL: https://github.com/apache/doris/pull/61182#discussion_r2916501673
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -166,11 +168,12 @@ private RecordWithMeta buildRecordResponse(
}
// Process data messages
- List<String> serializedRecords =
+ DeserializeResult result =
sourceReader.deserialize(fetchRecord.getConfig(),
element);
- if (!CollectionUtils.isEmpty(serializedRecords)) {
+ if (result.getType() == DeserializeResult.Type.DML
Review Comment:
`fetchRecords` currently only appends results when `DeserializeResult.Type`
is `DML`. If a deserializer returns `SCHEMA_CHANGE` (e.g., PG schema-change
detection), the triggering record’s `records` payload is silently dropped here,
which can lead to missing data and/or the fetch loop timing out without ever
returning data. Consider either (a) treating `SCHEMA_CHANGE` as data for
`fetchRecords` (at least append `result.getRecords()`), or (b) ensuring
schema-change detection is disabled / never returns `SCHEMA_CHANGE` on the
`/api/fetchRecords` path (since this path doesn’t execute DDLs and doesn’t
provide `Constants.DORIS_TARGET_DB`).
```suggestion
if ((result.getType() == DeserializeResult.Type.DML
|| result.getType() ==
DeserializeResult.Type.SCHEMA_CHANGE)
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/HttpUtil.java:
##########
@@ -52,4 +52,8 @@ protected boolean isRedirectable(String method) {
.addInterceptorLast(new RequestContent(true))
.build();
}
+
+ public static String getAuthHeader() {
+ return "Basic YWRtaW46";
+ }
Review Comment:
`getAuthHeader()` returns a hard-coded Basic auth value (`Basic YWRtaW46`),
which embeds a fixed credential and is reused across FE APIs (stream load +
schema change). This should be derived from configured user/password (or
avoided entirely if the `token` header is the real auth mechanism) to prevent
shipping credentials in code and to support non-admin users.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java:
##########
@@ -0,0 +1,66 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.cdcclient.source.deserialize;
+
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+
+import org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils;
+import org.apache.flink.cdc.debezium.history.FlinkJsonTableChangeSerializer;
+import org.apache.kafka.connect.source.SourceRecord;
+
+import java.io.IOException;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MySQL-specific deserializer that handles DDL schema change events.
+ *
+ * <p>When a schema change event is detected, it parses the HistoryRecord,
computes the diff against
+ * stored tableSchemas, generates Doris ALTER TABLE SQL, and returns a
SCHEMA_CHANGE result.
+ */
+public class MySqlDebeziumJsonDeserializer extends DebeziumJsonDeserializer {
+ private static final long serialVersionUID = 1L;
+ private static final Logger LOG =
LoggerFactory.getLogger(MySqlDebeziumJsonDeserializer.class);
+ private static final FlinkJsonTableChangeSerializer
TABLE_CHANGE_SERIALIZER =
+ new FlinkJsonTableChangeSerializer();
+
+ private String targetDb;
Review Comment:
This class currently defines `LOG`, `TABLE_CHANGE_SERIALIZER`, and
`targetDb`, but none of them are used (and `handleSchemaChangeEvent` is a stub
that always returns EMPTY). If MySQL schema change handling is intentionally
unimplemented for now, consider removing these unused members/imports (or
wiring the implementation) to avoid dead code and potential style/check
failures.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
