Copilot commented on code in PR #67851:
URL: https://github.com/apache/doris/pull/67851#discussion_r3988270530
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -452,6 +452,9 @@ public CompletableFuture<Void>
writeRecordsAsync(WriteRecordRequest writeRecordR
* <p>Heartbeat events will carry the latest offset.
*/
public void writeRecords(WriteRecordRequest writeRecordRequest) throws
Exception {
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(writeRecordRequest.getDorisUser()),
+ "Missing dorisUser; FE must send the Doris job creator's user
name");
Review Comment:
Making `dorisUser` mandatory breaks mixed-version operation.
`WriteRecordRequest` already documents that an old FE may omit newer fields
(`reuseReader` at lines 37-38), so after upgrading the cdc_client before the
FE, every write is accepted asynchronously and then fails here before reading
any records. Token-authenticated stream load explicitly supports user-less
internal loads in `StreamLoadHandler.java:146-149`; keep this field optional
for old senders and only add the Basic user header when it is nonblank.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeManager.java:
##########
@@ -38,8 +38,8 @@
public class SchemaChangeManager {
private static final Logger LOG =
LoggerFactory.getLogger(SchemaChangeManager.class);
- private static final String SCHEMA_CHANGE_API =
"http://%s/api/query/default_cluster/%s";;
- private static final String TABLE_SCHEMA_API =
"http://%s/api/%s/%s/_schema";;
+ private static final String SCHEMA_CHANGE_API =
"http://%s/api/streaming/schema_change";;
+ private static final String TABLE_SCHEMA_API =
"http://%s/api/streaming/%s/%s/_schema";;
Review Comment:
These new routes are used unconditionally, but an older master FE does not
expose them. In a BE/cdc-client-first rolling upgrade, the next existing CDC
schema event receives 404 responses and fails even when the admin password is
unchanged. Preserve the legacy route as a capability/404 fallback for old FEs
(with legacy authentication only on that fallback), or otherwise negotiate FE
support before selecting these endpoints.
##########
fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/StreamingJobAction.java:
##########
@@ -50,14 +78,69 @@ public Object reportTaskFailure(@RequestBody
TaskFailureRequest failureRequest,
return failTask(failureRequest);
}
+ @RequestMapping(path = "/api/streaming/schema_change", method =
RequestMethod.POST)
+ public Object executeSchemaChange(@RequestBody Map<String, String> body,
HttpServletRequest request) {
+ checkAuth(request);
+ if (!Env.getCurrentEnv().isMaster()) {
+ return ResponseEntityBuilder.okWithCommonError("Schema change must
be executed on the master FE");
+ }
+ String stmt = body.get("stmt");
+ if (Strings.isNullOrEmpty(stmt)) {
+ return ResponseEntityBuilder.badRequest("Missing statement request
body");
+ }
+
+ ConnectContext ctx = createJobContext(request);
+ try (AutoCloseConnectContext ignored = new
AutoCloseConnectContext(ctx)) {
+ StmtExecutor executor = new StmtExecutor(ctx, stmt);
+ executor.execute();
+ if (ctx.getState().getStateType() ==
QueryState.MysqlStateType.ERR) {
+ return
ResponseEntityBuilder.okWithCommonError(ctx.getState().getErrorMessage());
+ }
+ return ResponseEntityBuilder.ok();
+ } catch (Exception e) {
+ LOG.warn("Failed to execute schema change", e);
+ return ResponseEntityBuilder.okWithCommonError(e.getMessage());
+ }
+ }
+
private void checkAuth(HttpServletRequest request) {
String authToken = request.getHeader("token");
if (Strings.isNullOrEmpty(authToken)) {
throw new UnauthorizedException("Miss token");
}
if (!checkClusterToken(authToken)) {
- throw new UnauthorizedException("Invalid token: " + authToken);
+ throw new UnauthorizedException("Invalid token");
+ }
+ }
+
+ // Call only after validating the internal token. The caller owns the
context's scope.
+ private static ConnectContext createJobContext(HttpServletRequest request)
{
+ String jobIdHeader = request.getHeader("jobId");
+ if (Strings.isNullOrEmpty(jobIdHeader)) {
+ throw new BadRequestException("Missing jobId header; CDC client
must send the streaming job ID");
+ }
+ long jobId;
+ try {
+ jobId = Long.parseLong(jobIdHeader);
+ } catch (NumberFormatException e) {
+ throw new BadRequestException("Invalid jobId header: " +
jobIdHeader);
+ }
+ AbstractJob job = Env.getCurrentEnv().getJobManager().getJob(jobId);
+ if (!(job instanceof StreamingInsertJob)) {
+ throw new BadRequestException("Job " + jobId + " is not a
streaming job or does not exist");
Review Comment:
The token is cluster-wide and resolving only `jobId` does not bind this
request to the current task. A timed-out task can keep running while FE
installs a successor (possibly on another BE), then issue DDL that bypasses the
stale-task checks already present in commit-offset and failure-report paths;
this is especially unsafe after a paused job is altered before resume. Send
`taskId` with schema requests and reject it unless it matches the job's current
running task and status before creating the context.
--
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]
