github-actions[bot] commented on code in PR #64878:
URL: https://github.com/apache/doris/pull/64878#discussion_r3601111001
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -885,6 +853,117 @@ private void modifyPropertiesInternal(Map<String, String>
jobProperties,
this.id, jobProperties, dataSourceProperties);
}
+ private PreparedKafkaProperties
prepareDataSourceProperties(KafkaDataSourceProperties dataSourceProperties,
+ boolean isReplay) throws UserException {
+ if (dataSourceProperties == null) {
+ return null;
+ }
+
+ KafkaDataSourceSnapshot dataSourceSnapshot =
stageDataSourceProperties(dataSourceProperties);
+ List<Pair<Integer, Long>> kafkaPartitionOffsets = Lists.newArrayList();
+ if
(MapUtils.isNotEmpty(dataSourceProperties.getOriginalDataSourceProperties())) {
+ kafkaPartitionOffsets =
dataSourceProperties.getKafkaPartitionOffsets();
+ }
+ if (!kafkaPartitionOffsets.isEmpty()
Review Comment:
[P1] Validate numeric-offset partitions against the staged topic. For an
unpinned topic switch this condition skips the old-progress check, and
resolveOffsets() cannot replace it: KafkaUtil.getRealOffsets() returns an
all-numeric list without contacting Kafka. ALTER ... kafka_topic=new,
kafka_partitions=999, kafka_offsets=0 therefore succeeds and journals bogus
progress even when new has only partition 0; dynamic discovery later ignores
999 and starts real partitions from the default. Fetch the staged topic's
partition metadata for every requested ID, including all-numeric and mixed
lists, before the Cloud reset or local apply.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -885,6 +853,117 @@ private void modifyPropertiesInternal(Map<String, String>
jobProperties,
this.id, jobProperties, dataSourceProperties);
}
+ private PreparedKafkaProperties
prepareDataSourceProperties(KafkaDataSourceProperties dataSourceProperties,
+ boolean isReplay) throws UserException {
+ if (dataSourceProperties == null) {
+ return null;
+ }
+
+ KafkaDataSourceSnapshot dataSourceSnapshot =
stageDataSourceProperties(dataSourceProperties);
+ List<Pair<Integer, Long>> kafkaPartitionOffsets = Lists.newArrayList();
+ if
(MapUtils.isNotEmpty(dataSourceProperties.getOriginalDataSourceProperties())) {
+ kafkaPartitionOffsets =
dataSourceProperties.getKafkaPartitionOffsets();
+ }
+ if (!kafkaPartitionOffsets.isEmpty()
+ && (!dataSourceSnapshot.resetProgress ||
CollectionUtils.isNotEmpty(customKafkaPartitions))) {
+ ((KafkaProgress) progress).checkPartitions(kafkaPartitionOffsets);
+ }
+
+ if (!isReplay) {
+ kafkaPartitionOffsets = resolveOffsets(dataSourceProperties,
dataSourceSnapshot);
+ resetCloudProgressIfNeeded(dataSourceSnapshot,
kafkaPartitionOffsets);
Review Comment:
[P1] Do not commit Cloud progress during preparation. This RPC runs before
the new target is re-resolved/replanned and before local apply or the EditLog
write. In a combined target/source ALTER, another session can drop the analyzed
target or make its schema incompatible; the reset commits, final validation
then throws, and the FE job/journal stay unchanged while MetaService contains
the new offsets. A later progress refresh or failover can import them into the
old job and skip or replay records. Use a durable two-phase/conditional
protocol or compensation so a rejected ALTER cannot change MetaService.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java:
##########
@@ -166,13 +193,21 @@ public void validate(ConnectContext ctx) throws
UserException {
// check load properties
RoutineLoadJob job = Env.getCurrentEnv().getRoutineLoadManager()
.getJob(getDbName(), getJobName());
- this.routineLoadDesc = CreateRoutineLoadInfo.checkLoadProperties(ctx,
loadPropertyMap,
- job.getDbFullName(), job.getTableName(), job.isMultiTable(),
job.getMergeType());
+ if (MapUtils.isNotEmpty(loadPropertyMap)) {
+ this.routineLoadDesc =
CreateRoutineLoadInfo.checkLoadProperties(ctx, loadPropertyMap,
+ job.getDbFullName(), job.getTableName(),
job.isMultiTable(), job.getMergeType());
Review Comment:
[P1] Bind load-clause validation and application to the current target. Two
concurrent ALTERs can bypass the same-statement exclusion: T1 validates WHERE
k1 against target A here; T2 switches the paused job to incompatible B; then T1
completes and RoutineLoadManager installs its A-derived RoutineLoadDesc after
modifyProperties() has released the job lock. T1 succeeds but the job fails
planning when resumed on B. This is distinct from the existing
privilege/existence race because authorization may correctly use B. Revalidate
and apply the descriptor in the same target-version/job-lock and journal
boundary.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -907,7 +986,11 @@ private void
resetCloudProgress(Cloud.ResetRLProgressRequest.Builder builder) th
@Override
public void replayModifyProperties(AlterRoutineLoadJobOperationLog log) {
try {
- modifyPropertiesInternal(log.getJobProperties(),
(KafkaDataSourceProperties) log.getDataSourceProperties());
+ modifyPropertiesInternal(log.getJobProperties(),
+ (KafkaDataSourceProperties) log.getDataSourceProperties());
+ if (log.getTargetTableId() != 0) {
Review Comment:
[P1] Make replay independent of follower-local progress before applying the
target. In Cloud mode a follower's KafkaProgress can lag the MetaService-owned
master state. A combined target plus same-topic offset record for a dynamically
discovered partition can therefore pass on the master but make this replay call
throw checkPartitions() on the follower; the catch below consumes the journal
record before the newly added tableId assignment runs. A later failover retains
the old target. Replay the already validated prepared transition without
state-dependent membership checks, and cover same-version Cloud failover with
stale follower progress.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java:
##########
@@ -675,25 +674,11 @@ private Map<String, String>
getMaskedCustomProperties(String keyPrefix) {
}
@Override
- public void modifyProperties(AlterRoutineLoadCommand command) throws
UserException {
+ protected PreparedAlter prepareAlter(AlterRoutineLoadCommand command) {
Map<String, String> jobProperties = command.getAnalyzedJobProperties();
KinesisDataSourceProperties dataSourceProperties =
(KinesisDataSourceProperties)
command.getDataSourceProperties();
-
- writeLock();
- try {
- if (getState() != JobState.PAUSED) {
- throw new DdlException("Only supports modification of PAUSED
jobs");
- }
-
- modifyPropertiesInternal(jobProperties, dataSourceProperties);
-
- AlterRoutineLoadJobOperationLog log = new
AlterRoutineLoadJobOperationLog(this.id,
- jobProperties, dataSourceProperties);
- Env.getCurrentEnv().getEditLog().logAlterRoutineLoadJob(log);
- } finally {
- writeUnlock();
- }
+ return () -> modifyPropertiesInternal(jobProperties,
dataSourceProperties);
Review Comment:
[P1] Preserve the stored Kinesis default when this ALTER delta omits one.
KinesisDataSourceProperties still calls analyzeKinesisDefaultPositionProperty()
for every ALTER without explicit positions, which inserts
kinesis_default_pos=LATEST into the delta. This closure feeds that synthetic
value to customProperties.putAll(), so a stream/region/endpoint/custom-only
ALTER overwrites an existing TRIM_HORIZON; on a stream switch, initial shard
discovery then starts at LATEST and skips every record already present. Mirror
Kafka's ALTER early return when neither shards nor a default was supplied, and
cover live/replay/image preservation.
--
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]
