github-actions[bot] commented on code in PR #68079:
URL: https://github.com/apache/doris/pull/68079#discussion_r4025380542
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -1093,22 +1093,31 @@ public void replayOnUpdated(StreamingInsertJob
replayJob) {
*/
private void modifyPropertiesInternal(Map<String, String> inputProperties)
throws AnalysisException, JobException {
StreamingJobProperties inputStreamProps = new
StreamingJobProperties(inputProperties);
+ boolean reloadSourceSchema = Boolean.parseBoolean(
+
inputProperties.get(StreamingJobProperties.RELOAD_SOURCE_SCHEMA_PROPERTY));
if (StringUtils.isNotEmpty(inputStreamProps.getOffsetProperty())) {
Offset offset =
validateOffset(inputStreamProps.getOffsetProperty());
this.offsetProvider.updateOffset(offset);
this.offsetProvider.resetLag();
+ if (reloadSourceSchema) {
+ this.offsetProvider.resetSourceSchema();
Review Comment:
[P1] Make the offset/schema reset failure-atomic — These provider mutations
precede later fallible work. In cloud mode, an RPC failure returns before the
property update/edit log while this leader keeps the unjournaled new offset and
cleared schema; a post-commit timeout can also leave MS new while restart
restores the old blob. Even non-cloud combined ALTER can fail later during
source-certificate conversion with the same unjournaled mutation. Stage changes
until all validation/reset work succeeds, and reconcile ambiguous remote
outcomes (or persist an intent).
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -802,11 +820,41 @@ public void closeJobStreamLoad(String jobId) {
}
}
- private String extractTable(SourceRecord record) {
+ private static String extractTable(SourceRecord record) {
Struct value = (Struct) record.value();
return value.getStruct(Envelope.FieldName.SOURCE).getString("table");
}
+ static String formatSourceRecordFailure(
Review Comment:
[P1] Apply the cdc-client formatter to this method — Both exact-head `Build
Cdc Client` and `Test Cdc Client` jobs stop in `spotless:check` on this
signature and the root-cause conditional below, so none of the new CDC code or
unit tests is compiled or run. Apply the repository's google-java-format output
and rerun both checks.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -1093,22 +1093,31 @@ public void replayOnUpdated(StreamingInsertJob
replayJob) {
*/
private void modifyPropertiesInternal(Map<String, String> inputProperties)
throws AnalysisException, JobException {
StreamingJobProperties inputStreamProps = new
StreamingJobProperties(inputProperties);
+ boolean reloadSourceSchema = Boolean.parseBoolean(
+
inputProperties.get(StreamingJobProperties.RELOAD_SOURCE_SCHEMA_PROPERTY));
if (StringUtils.isNotEmpty(inputStreamProps.getOffsetProperty())) {
Offset offset =
validateOffset(inputStreamProps.getOffsetProperty());
this.offsetProvider.updateOffset(offset);
this.offsetProvider.resetLag();
+ if (reloadSourceSchema) {
+ this.offsetProvider.resetSourceSchema();
+ }
this.offsetProviderPersist = offsetProvider.getPersistInfo();
log.info("modifyPropertiesInternal: offset updated to {}, job {}",
inputStreamProps.getOffsetProperty(), getJobId());
if (Config.isCloudMode()) {
resetCloudProgress(offset);
}
+ } else if (reloadSourceSchema) {
+ this.offsetProvider.resetSourceSchema();
Review Comment:
[P1] Serialize this one-shot reset with resume and commits — The PAUSED
check occurs earlier in `AlterJobCommand.validate()`, while this mutation takes
no job lock or status recheck. Auto-resume can dispatch a task with the old
schema between those steps; its later `commitOffset()` then writes that schema
back after this reset succeeds. Recheck PAUSED and mutate schema/offset/rebuild
state in the same critical section that fences scheduler transitions and commit
callbacks.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -802,11 +820,41 @@ public void closeJobStreamLoad(String jobId) {
}
}
- private String extractTable(SourceRecord record) {
+ private static String extractTable(SourceRecord record) {
Struct value = (Struct) record.value();
return value.getStruct(Envelope.FieldName.SOURCE).getString("table");
}
+ static String formatSourceRecordFailure(
+ String action, SourceRecord record, Throwable failure) {
+ try {
+ StringBuilder message = new StringBuilder(action);
+ if (record.value() instanceof Struct) {
Review Comment:
[P2] Extract the table from schema-change records — This formatter only
recognizes a normal Debezium DML envelope. The failures passed here are
MySQL/OceanBase history records or PostgreSQL `PostgresSchemaRecord`s, so the
new error omits `Source table:` for the very DDL failures it targets (and the
Doris SQL may name a mapped target instead). Handle those schema-record shapes
and add focused coverage for all supported sources.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -1093,22 +1093,31 @@ public void replayOnUpdated(StreamingInsertJob
replayJob) {
*/
private void modifyPropertiesInternal(Map<String, String> inputProperties)
throws AnalysisException, JobException {
StreamingJobProperties inputStreamProps = new
StreamingJobProperties(inputProperties);
+ boolean reloadSourceSchema = Boolean.parseBoolean(
+
inputProperties.get(StreamingJobProperties.RELOAD_SOURCE_SCHEMA_PROPERTY));
if (StringUtils.isNotEmpty(inputStreamProps.getOffsetProperty())) {
Offset offset =
validateOffset(inputStreamProps.getOffsetProperty());
this.offsetProvider.updateOffset(offset);
this.offsetProvider.resetLag();
+ if (reloadSourceSchema) {
+ this.offsetProvider.resetSourceSchema();
+ }
this.offsetProviderPersist = offsetProvider.getPersistInfo();
log.info("modifyPropertiesInternal: offset updated to {}, job {}",
inputStreamProps.getOffsetProperty(), getJobId());
if (Config.isCloudMode()) {
resetCloudProgress(offset);
}
+ } else if (reloadSourceSchema) {
+ this.offsetProvider.resetSourceSchema();
+ this.offsetProviderPersist = offsetProvider.getPersistInfo();
Review Comment:
[P1] Force a reader rebuild when reloading the source schema — A job can be
PAUSED without setting `needRebuildReader` (for example after data-quality,
metadata-fetch, or split-advance failure). This branch clears only the FE blob;
resume then sends `rebuildReader=false` with no schemas, so the cdc client
retains the old reader and `loadTableSchemasFromJson(null)` leaves its cached
schema untouched. The ALTER therefore reports success but resumes against the
stale schema. Mark the reader for rebuild/release as part of this reset and
cover a non-task-failure pause.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -604,21 +608,35 @@ public void writeRecords(WriteRecordRequest
writeRecordRequest) throws Exception
}
// Process data messages
- DeserializeResult result =
- sourceReader.deserialize(deserializeContext,
element);
+ DeserializeResult result;
+ try {
+ result = sourceReader.deserialize(deserializeContext,
element);
+ } catch (Exception e) {
+ throw new SourceRecordProcessingException(
+ formatSourceRecordFailure(
+ "Failed to deserialize source record",
element, e),
+ e);
+ }
if (result.getType() ==
DeserializeResult.Type.SCHEMA_CHANGE) {
// Flush pending data before DDL
batchStreamLoad.forceFlush();
if
(!CollectionUtils.isEmpty(result.getSchemaChanges())) {
ddlCount += result.getSchemaChanges().size();
}
- SchemaChangeManager.executeChanges(
- feAddr,
- targetDb,
- token,
- writeRecordRequest.getJobId(),
- result.getSchemaChanges());
+ try {
+ SchemaChangeManager.executeChanges(
+ feAddr,
+ targetDb,
+ token,
+ writeRecordRequest.getJobId(),
+ result.getSchemaChanges());
+ } catch (Exception e) {
+ throw new SourceRecordProcessingException(
Review Comment:
[P1] Do not skip unreported operations from this schema record — One source
DDL can produce several Doris operations, but `executeChanges()` stops at the
first failure and this error exposes only that operation. MySQL/OceanBase
resume after the whole record; PostgreSQL replays its Relation, but the rebuilt
reader first discovers the source's final schema, so the event produces no
diff. With `reload_source_schema`, either path can permanently lose the
unexecuted suffix. Report every operation the administrator must apply (or
retain a baseline/position that safely replays it), and test multi-operation
recovery for both restart models.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeManager.java:
##########
@@ -103,7 +103,9 @@ public static void execute(
} catch (IOException schemaFailure) {
ddlFailure.addSuppressed(schemaFailure);
}
- throw ddlFailure;
+ throw new IOException(
Review Comment:
[P1] Keep the diagnostics and exception contract consistent — In the
schema-query-failure branch, the original DDL message and suppressed query
error are attached to `ddlFailure`; this fresh wrapper exposes neither at the
top level. The unchanged `schemaQueryFailureKeepsOriginalDdlFailure` test
therefore fails both its message and suppressed-exception assertions. Preserve
equivalent diagnostics and align the direct API/test contract while adding SQL
context. Also apply the repository formatter here: both exact-head CDC jobs
currently stop at Spotless on this statement before tests run.
--
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]
