github-actions[bot] commented on code in PR #64878:
URL: https://github.com/apache/doris/pull/64878#discussion_r3635596763
##########
fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java:
##########
@@ -237,6 +241,101 @@ public void
testUpdateLagRebuildsConvertedPropertiesAfterReplay() throws UserExc
}
}
+ @Test
+ public void testModifyTargetTableWithJobAndDataSourceProperties() throws
Exception {
+ KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L,
"kafka_routine_load_job", 1L,
+ 101L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN);
+ Deencapsulation.setField(routineLoadJob, "state",
RoutineLoadJob.JobState.PAUSED);
+ KafkaProgress progress = new KafkaProgress(Maps.newHashMap());
+ Deencapsulation.setField(routineLoadJob, "progress", progress);
+
Review Comment:
[P1] Preserve the existing Kafka default in this custom-only combined ALTER.
Although this statement supplies only client.id,
KafkaDataSourceProperties.analyze() falls through to
analyzeKafkaDefaultOffsetProperty(), which synthesizes
kafka_default_offsets=OFFSET_END; modifyPropertiesInternal() then merges that
value into the live job before assigning and journaling the new target. A job
created with OFFSET_BEGINNING can therefore execute the advertised SET TARGET
TABLE ... FROM KAFKA("property.client.id"=...) form and silently make every
later-discovered partition start at the end, skipping its backlog. This test
masks the overwrite because the job starts with the implicit END default and
never asserts it. Keep the default absent from an ALTER delta unless it was
explicitly supplied (or needed for explicit partitions), and cover
live/replay/image preservation from OFFSET_BEGINNING.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java:
##########
@@ -687,9 +687,12 @@ public void modifyProperties(AlterRoutineLoadCommand
command) throws UserExcepti
}
modifyPropertiesInternal(jobProperties, dataSourceProperties);
+ if (command.hasTargetTable()) {
+ this.tableId = command.getTargetTableId();
+ }
AlterRoutineLoadJobOperationLog log = new
AlterRoutineLoadJobOperationLog(this.id,
- jobProperties, dataSourceProperties);
+ jobProperties, dataSourceProperties,
command.getTargetTableId());
Review Comment:
[P1] Validate explicit shards against the staged stream before journaling
this transition. When kinesis_stream is supplied, modifyPropertiesInternal()
clears progress, copies the requested IDs into customKinesisShards, and skips
checkShards() because resetProgress is true. On resume that nonempty custom
list also makes refreshKafkaPartitions() skip ListShards, so SET TARGET TABLE
... FROM
KINESIS("kinesis_stream"="stream-b","kinesis_shards"="missing","kinesis_shards_pos"="LATEST")
succeeds and persists a definition that will schedule a nonexistent shard.
Resolve the effective endpoint/region/stream and fetch its actual shards
outside the job lock, then recheck the job-definition generation and reject
missing IDs before publishing the target/source delta.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java:
##########
@@ -687,9 +687,12 @@ public void modifyProperties(AlterRoutineLoadCommand
command) throws UserExcepti
}
modifyPropertiesInternal(jobProperties, dataSourceProperties);
+ if (command.hasTargetTable()) {
+ this.tableId = command.getTargetTableId();
Review Comment:
[P1] Reset Kinesis state when the region changes. The source mutation below
sets resetProgress only for a new stream name; an ALTER that keeps the same
name but changes aws.region commits this new target and region while retaining
progress, custom/open/closed shard IDs, and lag from the old physical stream.
Pinned jobs then skip ListShards completely, and dynamic jobs can reuse an old
sequence number when the new region exposes the same standard shard ID,
producing an invalid starting position or repeated failures. Treat a region
change as a source-identity change just like a stream change (or require
validated explicit positions), and cover both pinned and dynamic jobs.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java:
##########
@@ -783,6 +786,9 @@ public void
replayModifyProperties(AlterRoutineLoadJobOperationLog log) {
try {
modifyPropertiesInternal(log.getJobProperties(),
(KinesisDataSourceProperties)
log.getDataSourceProperties());
+ if (log.getTargetTableId() != 0) {
Review Comment:
[P1] Do not make Kinesis replay depend on follower-local shard discovery.
Dynamic discovery adds shards to the leader's progress without journaling that
discovery, so the leader can discover shard-0, pause before any task commit,
and successfully journal an explicit shard position plus this target. A
follower or older image with empty progress then replays the same record,
checkShards() throws, this catch consumes the journal entry, and the tableId
assignment here is skipped. The new replay test cannot exercise this because
its captured record has no explicit shard positions (and it also pre-seeds
shard-0). Replay a fully prepared leader-validated transition without
consulting local progress (or persist the required canonical state in the same
record), and add the empty-progress failover/image case.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -782,9 +782,12 @@ public void modifyProperties(AlterRoutineLoadCommand
command) throws UserExcepti
}
modifyPropertiesInternal(jobProperties, dataSourceProperties);
+ if (command.hasTargetTable()) {
+ tableId = command.getTargetTableId();
Review Comment:
[P1] Re-seed progress when a broker-list change selects a different Kafka
cluster. With the topic unchanged, modifyPropertiesInternal() only replaces
brokerList; it retains progress, custom/current partitions, and the
latest-offset cache. Pinned jobs then skip metadata discovery, and a dynamic
job whose new cluster has the same numeric partition IDs appears unchanged, so
tasks combine the new brokers and target here with offsets from the old
cluster. Switching cluster A at offset 1000 to cluster B can therefore skip B's
first 1000 records or pause with OFFSET_OUT_OF_RANGE. Resolve the cluster ID
and any requested offsets against the effective new brokers outside the job
lock; if the cluster changed, require those validated offsets or reset/reseed
all progress and partition caches, persist that prepared decision for replay,
and recheck the job generation before committing.
--
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]
