github-actions[bot] commented on code in PR #66089:
URL: https://github.com/apache/doris/pull/66089#discussion_r3664855963
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/kafka/KafkaUtil.java:
##########
@@ -235,34 +243,8 @@ private static InternalService.PProxyResult
getInfoRequest(InternalService.PProx
try {
while (retryTimes < 3) {
- List<Long> backendIds = new ArrayList<>();
- for (Long beId :
Env.getCurrentSystemInfo().getAllBackendIds(true)) {
- Backend backend =
Env.getCurrentSystemInfo().getBackend(beId);
- if (isBackendAvailableForMetaRequest(backend)
- && !failedBeIds.contains(beId)
- &&
!Env.getCurrentEnv().getRoutineLoadManager().isInBlacklist(beId)) {
- backendIds.add(beId);
- }
- }
- // If there are no available backends, utilize the blacklist.
- // Special scenarios include:
- // 1. A specific job that connects to Kafka may time out for
topic config or network error,
- // leaving only one backend operational.
- // 2. If that sole backend is decommissioned, the
aliveBackends list becomes empty.
- // Hence, in such cases, it's essential to rely on the
blacklist to obtain meta information.
- if (backendIds.isEmpty()) {
- Map<Long, Long> blacklist =
Env.getCurrentEnv().getRoutineLoadManager().getBlacklist();
- for (Long beId : blacklist.keySet()) {
- Backend backend =
Env.getCurrentSystemInfo().getBackend(beId);
- if (isBackendAvailableForMetaRequest(backend) &&
!failedBeIds.contains(beId)) {
- backendIds.add(beId);
- } else if (backend == null) {
- blacklist.remove(beId);
- LOG.warn("remove stale backend {} from routine
load blacklist when getting kafka meta",
- beId);
- }
- }
- }
+ List<Long> candidateBackendIds =
getBackendIdsForMetaRequest(computeGroupName);
+ List<Long> backendIds =
getAvailableBackendIdsForMetaRequest(candidateBackendIds, failedBeIds);
if (backendIds.isEmpty()) {
Review Comment:
[P2] Include the compute group in scoped failures
This candidate list is now scoped to `computeGroupName`, but an unknown,
renamed, suspended, or backend-less group still falls through to the global
message `no alive backends`. Other compute groups may be healthy, and the
public wrappers do not restore the missing group name, so the job's pause/error
reason no longer identifies the routing key that failed. Please include
`computeGroupName` in both terminal `getInfoRequest` failures (or centralize
their error construction), and ideally distinguish an unknown group from a
known group with no available backend, as the adjacent cloud backend selectors
already do.
##########
fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java:
##########
@@ -181,14 +181,16 @@ public void testDivideRoutineLoadJob() throws
UserException {
public void testUpdateLagRefreshesLatestOffsetCache() throws UserException
{
KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L,
"kafka_routine_load_job", 1L,
1L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN);
+ routineLoadJob.setCloudCluster("routine-load-compute-group");
Review Comment:
[P1] Enable cloud mode before asserting group propagation
Setting the job's `cloudCluster` field does not make `Config.isCloudMode()`
true. With the test defaults, `updateLag()` passes `null` as the last argument,
so this stub requiring `"routine-load-compute-group"` does not match and no
latest offsets are cached. FE UT now confirms the result: this test fails at
line 198 with `expected:<15> but was:<0>`. Please save, enable, and restore a
cloud-mode config (for example `Config.cloud_unique_id`) around this test so it
both passes in isolation and actually proves the new propagation branch.
--
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]
