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:
##########
@@ -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/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);
+
+ Map<String, String> sourceProperties = Maps.newHashMap();
+ sourceProperties.put("property.client.id", "target-switch");
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:
##########
@@ -783,6 +786,9 @@ public void
replayModifyProperties(AlterRoutineLoadJobOperationLog log) {
try {
modifyPropertiesInternal(log.getJobProperties(),
(KinesisDataSourceProperties)
log.getDataSourceProperties());
+ if (log.getTargetTableId() != 0) {
+ this.tableId = log.getTargetTableId();
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.
--
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]
