Copilot commented on code in PR #68270:
URL: https://github.com/apache/doris/pull/68270#discussion_r4056657815
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -511,6 +511,18 @@ public Offset validateOffset(String offsetStr) throws
AnalysisException {
public void alterJob(AlterJobCommand alterJobCommand) throws
AnalysisException, JobException {
List<String> logParts = new ArrayList<>();
+
+ validateComputeGroupProperty(alterJobCommand.getProperties());
+
+ // Validate the merged properties before mutating the job.
+ Map<String, String> mergedSourceProperties = null;
+ Map<String, String> newConvertedSourceProperties = null;
+ if (!alterJobCommand.getSourceProperties().isEmpty()) {
+ mergedSourceProperties = new HashMap<>(this.sourceProperties);
+
mergedSourceProperties.putAll(alterJobCommand.getSourceProperties());
+ newConvertedSourceProperties =
buildConvertedSourceProperties(mergedSourceProperties);
Review Comment:
`AlterJobCommand` validates only the partial map supplied by the user, but
this block never validates `mergedSourceProperties`. For example, an existing
MySQL job with `snapshot_parallelism=4` can alter only `server_id` to a two-ID
range: the partial validation defaults parallelism to 1, while the merged state
violates the required range width and fails later in cdc_client. Validate the
merged source map against `dataSourceType` before converting or assigning it.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -1405,6 +1412,10 @@ public void beforeAborted(TransactionState txnState)
throws TransactionException
@Override
public void afterCommitted(TransactionState txnState, boolean txnOperated)
throws UserException {
+ if (!txnOperated) {
+ writeUnlock();
+ return;
Review Comment:
Please add a regression test for `txnOperated=false` that runs the callback
after `beforeCommitted` and verifies the same job write lock can be acquired
again. This branch is the only protection against the cloud commit-failure
retry leak, but the current FE tests do not exercise this callback lifecycle.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -526,12 +526,42 @@ public Offset deserializeOffsetProperty(String offset) {
}
@Override
- public void validateAlterOffset(String offset) throws Exception {
- if (!DataSourceConfigValidator.isJsonOffset(offset)) {
- throw new AnalysisException(
- "ALTER JOB for CDC only supports JSON specific offset, "
- + "e.g. '{\"file\":\"binlog.000001\",\"pos\":\"154\"}' for
MySQL "
- + "or '{\"lsn\":\"12345678\"}' for PostgreSQL");
+ public void validateAlterOffset(String offset) throws AnalysisException {
+ JsonNode offsetNode;
+ try {
+ offsetNode = objectMapper.readTree(offset);
+ } catch (Exception e) {
+ throw new AnalysisException("ALTER JOB for CDC requires a JSON
specific offset");
+ }
+ if (offsetNode == null || !offsetNode.isObject()) {
+ throw new AnalysisException("ALTER JOB for CDC requires a JSON
specific offset");
+ }
Review Comment:
This changes the existing user-visible error text for non-JSON CDC offsets.
The MySQL and PostgreSQL regression suites still expect `ALTER JOB for CDC only
supports JSON specific offset` for `offset='initial'`, so those tests will fail
even though this PR does not intend to change the contract. Keep the previous
message in both rejection branches (or update all callers and tests
consistently).
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -1095,22 +1099,25 @@ private void modifyPropertiesInternal(Map<String,
String> inputProperties) throw
StreamingJobProperties inputStreamProps = new
StreamingJobProperties(inputProperties);
if (StringUtils.isNotEmpty(inputStreamProps.getOffsetProperty())) {
Offset offset =
validateOffset(inputStreamProps.getOffsetProperty());
+ if (Config.isCloudMode()) {
+ resetCloudProgress(offset);
+ }
this.offsetProvider.updateOffset(offset);
this.offsetProvider.resetLag();
this.offsetProviderPersist = offsetProvider.getPersistInfo();
log.info("modifyPropertiesInternal: offset updated to {}, job {}",
inputStreamProps.getOffsetProperty(), getJobId());
- if (Config.isCloudMode()) {
- resetCloudProgress(offset);
- }
}
if
(inputProperties.containsKey(StreamingJobProperties.COMPUTE_GROUP_PROPERTY)) {
this.cloudCluster =
inputProperties.get(StreamingJobProperties.COMPUTE_GROUP_PROPERTY);
offsetProvider.setCloudCluster(this.cloudCluster);
}
+ long oldMaxIntervalSecond = this.jobProperties.getMaxIntervalSecond();
this.properties.putAll(inputProperties);
this.jobProperties = new StreamingJobProperties(this.properties);
- recomputeDerivedFields();
+ if (oldMaxIntervalSecond != this.jobProperties.getMaxIntervalSecond())
{
+ recomputeDerivedFields();
Review Comment:
Please add a regression test for the false branch here.
`StreamingInsertJobCheckDataQualityTest` covers recomputation when
`max_interval` changes, but it does not alter an unrelated property and assert
that `sampleStartTime` and both sample counters stay unchanged—the preservation
behavior introduced by this branch can otherwise regress silently.
--
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]
