This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new c3865359edc branch-4.1: [fix](job) keep Kafka metadata requests in
compute group (#66866)
c3865359edc is described below
commit c3865359edc1645b05bf788ebd38bfd97a0a91c0
Author: hui lai <[email protected]>
AuthorDate: Thu Aug 20 09:22:11 2026 +0800
branch-4.1: [fix](job) keep Kafka metadata requests in compute group
(#66866)
pick #66089.
---
.../apache/doris/datasource/kafka/KafkaUtil.java | 121 +++++++++++------
.../load/routineload/KafkaRoutineLoadJob.java | 33 +++--
.../doris/datasource/kafka/KafkaUtilTest.java | 150 +++++++++++++++++++++
.../load/routineload/KafkaRoutineLoadJobTest.java | 66 +++++----
.../doris/load/routineload/RoutineLoadJobTest.java | 7 +-
5 files changed, 300 insertions(+), 77 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/kafka/KafkaUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/kafka/KafkaUtil.java
index 8562d0eaed3..ebdb9ea3929 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/kafka/KafkaUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/kafka/KafkaUtil.java
@@ -18,6 +18,7 @@
package org.apache.doris.datasource.kafka;
import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
import org.apache.doris.common.LoadException;
import org.apache.doris.common.Pair;
@@ -26,9 +27,11 @@ import org.apache.doris.metric.MetricRepo;
import org.apache.doris.proto.InternalService;
import org.apache.doris.rpc.BackendServiceProxy;
import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TStatusCode;
+import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -48,7 +51,7 @@ public class KafkaUtil {
private static final Logger LOG = LogManager.getLogger(KafkaUtil.class);
public static List<Integer> getAllKafkaPartitions(String brokerList,
String topic,
- Map<String, String> convertedCustomProperties) throws
UserException {
+ Map<String, String> convertedCustomProperties, String
computeGroupName) throws UserException {
try {
InternalService.PProxyRequest request =
InternalService.PProxyRequest.newBuilder().setKafkaMetaRequest(
InternalService.PKafkaMetaProxyRequest.newBuilder()
@@ -61,7 +64,7 @@ public class KafkaUtil {
)
)
).build();
- return getInfoRequest(request,
Config.max_get_kafka_meta_timeout_second)
+ return getInfoRequest(request,
Config.max_get_kafka_meta_timeout_second, computeGroupName)
.getKafkaMetaResult().getPartitionIdsList();
} catch (Exception e) {
throw new LoadException(
@@ -73,7 +76,8 @@ public class KafkaUtil {
// The input parameter "timestampOffsets" is <partition, timestamp>
// Tne return value is <partition, offset>
public static List<Pair<Integer, Long>> getOffsetsForTimes(String
brokerList, String topic,
- Map<String, String> convertedCustomProperties, List<Pair<Integer,
Long>> timestampOffsets)
+ Map<String, String> convertedCustomProperties, List<Pair<Integer,
Long>> timestampOffsets,
+ String computeGroupName)
throws LoadException {
if (LOG.isDebugEnabled()) {
LOG.debug("begin to get offsets for times of topic: {}, {}",
topic, timestampOffsets);
@@ -100,7 +104,8 @@ public class KafkaUtil {
InternalService.PProxyRequest request =
InternalService.PProxyRequest.newBuilder().setKafkaMetaRequest(
metaRequestBuilder).setTimeoutSecs(Config.max_get_kafka_meta_timeout_second).build();
- InternalService.PProxyResult result = getInfoRequest(request,
Config.max_get_kafka_meta_timeout_second);
+ InternalService.PProxyResult result = getInfoRequest(
+ request, Config.max_get_kafka_meta_timeout_second,
computeGroupName);
List<InternalService.PIntegerPair> pairs =
result.getPartitionOffsets().getOffsetTimesList();
List<Pair<Integer, Long>> partitionOffsets = Lists.newArrayList();
@@ -120,7 +125,8 @@ public class KafkaUtil {
public static List<Pair<Integer, Long>> getLatestOffsets(long jobId, UUID
taskId, String brokerList, String topic,
Map<String,
String> convertedCustomProperties,
- List<Integer>
partitionIds) throws LoadException {
+ List<Integer>
partitionIds, String computeGroupName)
+ throws
LoadException {
if (LOG.isDebugEnabled()) {
LOG.debug("begin to get latest offsets for partitions {} in topic:
{}, task {}, job {}",
partitionIds, topic, taskId, jobId);
@@ -145,7 +151,8 @@ public class KafkaUtil {
}
InternalService.PProxyRequest request =
InternalService.PProxyRequest.newBuilder().setKafkaMetaRequest(
metaRequestBuilder).setTimeoutSecs(Config.max_get_kafka_meta_timeout_second).build();
- InternalService.PProxyResult result = getInfoRequest(request,
Config.max_get_kafka_meta_timeout_second);
+ InternalService.PProxyResult result = getInfoRequest(
+ request, Config.max_get_kafka_meta_timeout_second,
computeGroupName);
List<InternalService.PIntegerPair> pairs =
result.getPartitionOffsets().getOffsetTimesList();
List<Pair<Integer, Long>> partitionOffsets = Lists.newArrayList();
@@ -166,7 +173,7 @@ public class KafkaUtil {
public static List<Pair<Integer, Long>> getRealOffsets(String brokerList,
String topic,
Map<String,
String> convertedCustomProperties,
-
List<Pair<Integer, Long>> offsets)
+
List<Pair<Integer, Long>> offsets, String computeGroupName)
throws
LoadException {
// filter values greater than 0 as these offsets is real offset
// only update offset like OFFSET_BEGINNING or OFFSET_END
@@ -205,7 +212,8 @@ public class KafkaUtil {
}
InternalService.PProxyRequest request =
InternalService.PProxyRequest.newBuilder().setKafkaMetaRequest(
metaRequestBuilder).setTimeoutSecs(Config.max_get_kafka_meta_timeout_second).build();
- InternalService.PProxyResult result = getInfoRequest(request,
Config.max_get_kafka_meta_timeout_second);
+ InternalService.PProxyResult result = getInfoRequest(
+ request, Config.max_get_kafka_meta_timeout_second,
computeGroupName);
List<InternalService.PIntegerPair> pairs =
result.getPartitionOffsets().getOffsetTimesList();
List<Pair<Integer, Long>> partitionOffsets = Lists.newArrayList();
@@ -222,8 +230,8 @@ public class KafkaUtil {
}
}
- private static InternalService.PProxyResult
getInfoRequest(InternalService.PProxyRequest request, int timeout)
- throws LoadException {
+ private static InternalService.PProxyResult
getInfoRequest(InternalService.PProxyRequest request, int timeout,
+ String computeGroupName) throws LoadException {
long startTime = System.currentTimeMillis();
int retryTimes = 0;
TNetworkAddress address = null;
@@ -235,40 +243,20 @@ public class KafkaUtil {
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;
+ try {
+ candidateBackendIds =
getBackendIdsForMetaRequest(computeGroupName);
+ } catch (LoadException e) {
+
MetricRepo.COUNTER_ROUTINE_LOAD_GET_META_FAIL_COUNT.increase(1L);
+ throw new
LoadException(getInfoFailureMessage(e.getDetailMessage(), computeGroupName));
}
+ List<Long> backendIds =
getAvailableBackendIdsForMetaRequest(candidateBackendIds, failedBeIds);
if (backendIds.isEmpty()) {
MetricRepo.COUNTER_ROUTINE_LOAD_GET_META_FAIL_COUNT.increase(1L);
if (failedBeIds.isEmpty()) {
errorMsg = "no alive backends";
}
- throw new LoadException("failed to get info: " + errorMsg
+ ",");
+ throw new LoadException(getInfoFailureMessage(errorMsg,
computeGroupName));
}
Collections.shuffle(backendIds);
long selectedBeId = backendIds.get(0);
@@ -307,7 +295,7 @@ public class KafkaUtil {
}
MetricRepo.COUNTER_ROUTINE_LOAD_GET_META_FAIL_COUNT.increase(1L);
- throw new LoadException("failed to get info: " + errorMsg + ",");
+ throw new LoadException(getInfoFailureMessage(errorMsg,
computeGroupName));
} finally {
// Ensure that not all BE added to the blacklist.
// For single request:
@@ -329,6 +317,61 @@ public class KafkaUtil {
}
}
+ static String getInfoFailureMessage(String errorMsg, String
computeGroupName) {
+ String computeGroupDetails = Strings.isNullOrEmpty(computeGroupName)
+ ? "" : " compute group: " + computeGroupName + ",";
+ return "failed to get info: " + errorMsg + "," + computeGroupDetails;
+ }
+
+ static List<Long> getAvailableBackendIdsForMetaRequest(
+ List<Long> candidateBackendIds, Set<Long> failedBeIds) {
+ List<Long> backendIds = new ArrayList<>();
+ for (Long beId : candidateBackendIds) {
+ 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 : candidateBackendIds) {
+ if (!blacklist.containsKey(beId)) {
+ continue;
+ }
+ 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);
+ }
+ }
+ }
+ return backendIds;
+ }
+
+ static List<Long> getBackendIdsForMetaRequest(String computeGroupName)
throws LoadException {
+ SystemInfoService systemInfoService = Env.getCurrentSystemInfo();
+ if (!Config.isCloudMode()) {
+ return systemInfoService.getAllBackendIds(true);
+ }
+ if (Strings.isNullOrEmpty(computeGroupName)) {
+ throw new LoadException("compute group is empty when getting kafka
meta");
+ }
+ return ((CloudSystemInfoService)
systemInfoService).getBackendsByClusterName(computeGroupName).stream()
+ .map(Backend::getId)
+ .collect(Collectors.toList());
+ }
+
private static boolean isBackendAvailableForMetaRequest(Backend backend) {
return backend != null && backend.isLoadAvailable()
&& !backend.isDecommissioned() && !backend.isDecommissioning();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
index cb563e725db..579f860b46e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
@@ -545,7 +545,12 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
private List<Integer> getAllKafkaPartitions() throws UserException {
convertCustomProperties(false);
- return KafkaUtil.getAllKafkaPartitions(brokerList, topic,
convertedCustomProperties);
+ String computeGroupName = getComputeGroupName();
+ return KafkaUtil.getAllKafkaPartitions(brokerList, topic,
convertedCustomProperties, computeGroupName);
+ }
+
+ private String getComputeGroupName() {
+ return Config.isCloudMode() ? getCloudCluster() : null;
}
public static KafkaRoutineLoadJob fromCreateInfo(CreateRoutineLoadInfo
info, ConnectContext ctx)
@@ -639,13 +644,14 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
for (Integer kafkaPartition : newPartitions) {
partitionOffsets.add(Pair.of(kafkaPartition, beginOffset));
}
+ String computeGroupName = getComputeGroupName();
try {
if (isOffsetForTimes()) {
partitionOffsets =
KafkaUtil.getOffsetsForTimes(this.brokerList,
- this.topic, convertedCustomProperties,
partitionOffsets);
+ this.topic, convertedCustomProperties,
partitionOffsets, computeGroupName);
} else {
partitionOffsets = KafkaUtil.getRealOffsets(this.brokerList,
- this.topic, convertedCustomProperties,
partitionOffsets);
+ this.topic, convertedCustomProperties,
partitionOffsets, computeGroupName);
}
} catch (LoadException e) {
LOG.warn(new LogBuilder(LogKey.ROUTINE_LOAD_JOB, id)
@@ -677,15 +683,18 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
List<Pair<Integer, Long>> kafkaPartitionOffsets =
kafkaDataSourceProperties.getKafkaPartitionOffsets();
boolean isForTimes = kafkaDataSourceProperties.isOffsetsForTimes();
+ String computeGroupName = getComputeGroupName();
if (isForTimes) {
// the offset is set by date time, we need to get the real offset
by time
kafkaPartitionOffsets =
KafkaUtil.getOffsetsForTimes(kafkaDataSourceProperties.getBrokerList(),
kafkaDataSourceProperties.getTopic(),
- convertedCustomProperties,
kafkaDataSourceProperties.getKafkaPartitionOffsets());
+ convertedCustomProperties,
kafkaDataSourceProperties.getKafkaPartitionOffsets(),
+ computeGroupName);
} else {
kafkaPartitionOffsets =
KafkaUtil.getRealOffsets(kafkaDataSourceProperties.getBrokerList(),
kafkaDataSourceProperties.getTopic(),
- convertedCustomProperties,
kafkaDataSourceProperties.getKafkaPartitionOffsets());
+ convertedCustomProperties,
kafkaDataSourceProperties.getKafkaPartitionOffsets(),
+ computeGroupName);
}
for (Pair<Integer, Long> partitionOffset : kafkaPartitionOffsets) {
@@ -783,11 +792,14 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
if (partitionOffsets.isEmpty()) {
return;
}
+ String computeGroupName = getComputeGroupName();
List<Pair<Integer, Long>> newOffsets;
if (dataSourceProperties.isOffsetsForTimes()) {
- newOffsets = KafkaUtil.getOffsetsForTimes(brokerList, topic,
convertedCustomProperties, partitionOffsets);
+ newOffsets = KafkaUtil.getOffsetsForTimes(
+ brokerList, topic, convertedCustomProperties,
partitionOffsets, computeGroupName);
} else {
- newOffsets = KafkaUtil.getRealOffsets(brokerList, topic,
convertedCustomProperties, partitionOffsets);
+ newOffsets = KafkaUtil.getRealOffsets(
+ brokerList, topic, convertedCustomProperties,
partitionOffsets, computeGroupName);
}
dataSourceProperties.setKafkaPartitionOffsets(newOffsets);
}
@@ -943,8 +955,10 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
} finally {
writeUnlock();
}
+ String computeGroupName = getComputeGroupName();
List<Pair<Integer, Long>> tmp = KafkaUtil.getLatestOffsets(id,
taskId, brokerListSnapshot,
- topicSnapshot, customPropertiesSnapshot,
Lists.newArrayList(partitionIdToOffset.keySet()));
+ topicSnapshot, customPropertiesSnapshot,
Lists.newArrayList(partitionIdToOffset.keySet()),
+ computeGroupName);
updateLatestOffsetsCache(tmp, taskId);
} catch (Exception e) {
// It needs to pause job when can not get partition meta.
@@ -1007,8 +1021,9 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
writeUnlock();
}
UUID taskId = UUID.randomUUID();
+ String computeGroupName = getComputeGroupName();
List<Pair<Integer, Long>> latestOffsets =
KafkaUtil.getLatestOffsets(id, taskId, brokerListSnapshot,
- topicSnapshot, customPropertiesSnapshot, partitionIds);
+ topicSnapshot, customPropertiesSnapshot, partitionIds,
computeGroupName);
updateLatestOffsetsCache(latestOffsets, taskId);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/kafka/KafkaUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/kafka/KafkaUtilTest.java
new file mode 100644
index 00000000000..1036ff93834
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/kafka/KafkaUtilTest.java
@@ -0,0 +1,150 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.kafka;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.LoadException;
+import org.apache.doris.load.routineload.RoutineLoadManager;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+
+public class KafkaUtilTest {
+ @Test
+ public void testGetInfoFailureMessageIncludesComputeGroup() {
+ Assert.assertEquals("failed to get info: no alive backends, compute
group: routine-load-compute-group,",
+ KafkaUtil.getInfoFailureMessage("no alive backends",
"routine-load-compute-group"));
+ Assert.assertEquals("failed to get info: no alive backends,",
+ KafkaUtil.getInfoFailureMessage("no alive backends", null));
+ }
+
+ @Test
+ public void testGetBackendIdsForMetaRequestUsesRoutineLoadComputeGroup()
throws Exception {
+ String originalCloudUniqueId = Config.cloud_unique_id;
+ Backend routineLoadBackend = new Backend(10001L, "127.0.0.1", 9050);
+ Backend otherComputeGroupBackend = new Backend(10002L, "127.0.0.2",
9050);
+ CloudSystemInfoService systemInfoService =
Mockito.mock(CloudSystemInfoService.class);
+
Mockito.when(systemInfoService.getBackendsByClusterName("routine-load-compute-group"))
+ .thenReturn(Collections.singletonList(routineLoadBackend));
+ Mockito.when(systemInfoService.getAllBackendIds(true))
+ .thenReturn(Arrays.asList(routineLoadBackend.getId(),
otherComputeGroupBackend.getId()));
+
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ Config.cloud_unique_id = "test-cloud";
+
envStatic.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
+ List<Long> backendIds =
KafkaUtil.getBackendIdsForMetaRequest("routine-load-compute-group");
+
+
Assert.assertEquals(Collections.singletonList(routineLoadBackend.getId()),
backendIds);
+
Mockito.verify(systemInfoService).getBackendsByClusterName("routine-load-compute-group");
+ Mockito.verify(systemInfoService,
Mockito.never()).getAllBackendIds(true);
+ } finally {
+ Config.cloud_unique_id = originalCloudUniqueId;
+ }
+ }
+
+ @Test
+ public void
testGetAvailableBackendIdsForMetaRequestKeepsBlacklistFallbackInComputeGroup() {
+ long routineLoadBackendId = 10001L;
+ long otherComputeGroupBackendId = 10002L;
+ Backend routineLoadBackend = mockAvailableBackend();
+ Backend otherComputeGroupBackend = mockAvailableBackend();
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getBackend(routineLoadBackendId)).thenReturn(routineLoadBackend);
+
Mockito.when(systemInfoService.getBackend(otherComputeGroupBackendId)).thenReturn(otherComputeGroupBackend);
+
+ RoutineLoadManager routineLoadManager =
Mockito.mock(RoutineLoadManager.class);
+
Mockito.when(routineLoadManager.isInBlacklist(routineLoadBackendId)).thenReturn(true);
+ Map<Long, Long> blacklist = new HashMap<>();
+ blacklist.put(routineLoadBackendId, 1L);
+ blacklist.put(otherComputeGroupBackendId, 1L);
+ Mockito.when(routineLoadManager.getBlacklist()).thenReturn(blacklist);
+ Env env = Mockito.mock(Env.class);
+
Mockito.when(env.getRoutineLoadManager()).thenReturn(routineLoadManager);
+
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+
envStatic.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+
+ List<Long> backendIds =
KafkaUtil.getAvailableBackendIdsForMetaRequest(
+ Collections.singletonList(routineLoadBackendId), new
HashSet<>());
+
+
Assert.assertEquals(Collections.singletonList(routineLoadBackendId),
backendIds);
+ Mockito.verify(systemInfoService,
Mockito.never()).getBackend(otherComputeGroupBackendId);
+ }
+ }
+
+ @Test
+ public void
testGetBackendIdsForMetaRequestRejectsMissingCloudComputeGroup() {
+ String originalCloudUniqueId = Config.cloud_unique_id;
+ CloudSystemInfoService systemInfoService =
Mockito.mock(CloudSystemInfoService.class);
+
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ Config.cloud_unique_id = "test-cloud";
+
envStatic.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
+ LoadException nullException = Assert.assertThrows(
+ LoadException.class, () ->
KafkaUtil.getBackendIdsForMetaRequest(null));
+ LoadException emptyException = Assert.assertThrows(
+ LoadException.class, () ->
KafkaUtil.getBackendIdsForMetaRequest(""));
+
+ Assert.assertEquals("compute group is empty when getting kafka
meta", nullException.getDetailMessage());
+ Assert.assertEquals("compute group is empty when getting kafka
meta", emptyException.getDetailMessage());
+ Mockito.verifyNoInteractions(systemInfoService);
+ } finally {
+ Config.cloud_unique_id = originalCloudUniqueId;
+ }
+ }
+
+ @Test
+ public void testGetBackendIdsForMetaRequestPreservesNonCloudSelection()
throws Exception {
+ String originalCloudUniqueId = Config.cloud_unique_id;
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+ List<Long> allBackendIds = Arrays.asList(10001L, 10002L);
+
Mockito.when(systemInfoService.getAllBackendIds(true)).thenReturn(allBackendIds);
+
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ Config.cloud_unique_id = "";
+
envStatic.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
+ Assert.assertEquals(allBackendIds,
KafkaUtil.getBackendIdsForMetaRequest(null));
+ Mockito.verify(systemInfoService).getAllBackendIds(true);
+ } finally {
+ Config.cloud_unique_id = originalCloudUniqueId;
+ }
+ }
+
+ private Backend mockAvailableBackend() {
+ Backend backend = Mockito.mock(Backend.class);
+ Mockito.when(backend.isLoadAvailable()).thenReturn(true);
+ return backend;
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
index 2565f26a15b..19ebc50ffbb 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
@@ -206,30 +206,38 @@ public class KafkaRoutineLoadJobTest {
@Test
public void testUpdateLagRefreshesLatestOffsetCache() throws UserException
{
- KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L,
"kafka_routine_load_job", 1L,
- 1L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN);
- Map<Integer, Long> partitionIdToOffset = Maps.newHashMap();
- partitionIdToOffset.put(1, 10L);
- partitionIdToOffset.put(2, 20L);
- Deencapsulation.setField(routineLoadJob, "progress", new
KafkaProgress(partitionIdToOffset));
-
- new MockUp<KafkaUtil>() {
- @Mock
- public List<Pair<Integer, Long>> getLatestOffsets(long jobId, UUID
taskId, String brokerList, String topic,
- Map<String,
String> convertedCustomProperties,
- List<Integer>
partitionIds) {
- Assert.assertEquals(1L, jobId);
- Assert.assertEquals("127.0.0.1:9020", brokerList);
- Assert.assertEquals("topic1", topic);
- Assert.assertTrue(partitionIds.contains(1));
- Assert.assertTrue(partitionIds.contains(2));
- return Lists.newArrayList(Pair.of(1, 15L), Pair.of(2, 30L));
- }
- };
-
- routineLoadJob.updateLag();
-
- Assert.assertEquals(15L, routineLoadJob.totalLag().longValue());
+ String originalCloudUniqueId = Config.cloud_unique_id;
+ try {
+ Config.cloud_unique_id = "test-cloud";
+ 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");
+ Map<Integer, Long> partitionIdToOffset = Maps.newHashMap();
+ partitionIdToOffset.put(1, 10L);
+ partitionIdToOffset.put(2, 20L);
+ Deencapsulation.setField(routineLoadJob, "progress", new
KafkaProgress(partitionIdToOffset));
+
+ new MockUp<KafkaUtil>() {
+ @Mock
+ public List<Pair<Integer, Long>> getLatestOffsets(long jobId,
UUID taskId, String brokerList,
+ String topic, Map<String, String>
convertedCustomProperties, List<Integer> partitionIds,
+ String cloudCluster) {
+ Assert.assertEquals(1L, jobId);
+ Assert.assertEquals("127.0.0.1:9020", brokerList);
+ Assert.assertEquals("topic1", topic);
+ Assert.assertTrue(partitionIds.contains(1));
+ Assert.assertTrue(partitionIds.contains(2));
+ Assert.assertEquals("routine-load-compute-group",
cloudCluster);
+ return Lists.newArrayList(Pair.of(1, 15L), Pair.of(2,
30L));
+ }
+ };
+
+ routineLoadJob.updateLag();
+
+ Assert.assertEquals(15L, routineLoadJob.totalLag().longValue());
+ } finally {
+ Config.cloud_unique_id = originalCloudUniqueId;
+ }
}
@Test
@@ -414,11 +422,12 @@ public class KafkaRoutineLoadJobTest {
@Mock
public List<Pair<Integer, Long>> getLatestOffsets(long jobId, UUID
taskId, String brokerList, String topic,
Map<String,
String> convertedCustomProperties,
- List<Integer>
partitionIds) {
+ List<Integer>
partitionIds, String cloudCluster) {
Assert.assertEquals("SASL_PLAINTEXT",
convertedCustomProperties.get("security.protocol"));
Assert.assertEquals("PLAIN",
convertedCustomProperties.get("sasl.mechanism"));
Assert.assertEquals(1, partitionIds.size());
Assert.assertTrue(partitionIds.contains(1));
+ Assert.assertNull(cloudCluster);
return Lists.newArrayList(Pair.of(1, 15L));
}
};
@@ -516,7 +525,8 @@ public class KafkaRoutineLoadJobTest {
new MockUp<KafkaUtil>() {
@Mock
public List<Integer> getAllKafkaPartitions(String brokerList,
String topic,
- Map<String, String> convertedCustomProperties) throws
UserException {
+ Map<String, String> convertedCustomProperties, String
cloudCluster) throws UserException {
+ Assert.assertNull(cloudCluster);
return Lists.newArrayList(1, 2, 3);
}
};
@@ -525,8 +535,10 @@ public class KafkaRoutineLoadJobTest {
@Mock
public List<Pair<Integer, Long>> getRealOffsets(String brokerList,
String topic,
Map<String,
String> convertedCustomProperties,
-
List<Pair<Integer, Long>> offsetFlags)
+
List<Pair<Integer, Long>> offsetFlags,
+ String
cloudCluster)
throws
LoadException {
+ Assert.assertNull(cloudCluster);
List<Pair<Integer, Long>> pairList = new ArrayList<>();
pairList.add(Pair.of(1, 0L));
pairList.add(Pair.of(2, 0L));
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
index cc6962560f8..4b60b956d9e 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
@@ -293,7 +293,8 @@ public class RoutineLoadJobTest {
new MockUp<KafkaUtil>() {
@Mock
public List<Integer> getAllKafkaPartitions(String brokerList,
String topic,
- Map<String, String> convertedCustomProperties) throws
UserException {
+ Map<String, String> convertedCustomProperties, String
cloudCluster) throws UserException {
+ Assert.assertNull(cloudCluster);
return Lists.newArrayList(1, 2, 3);
}
};
@@ -302,8 +303,10 @@ public class RoutineLoadJobTest {
@Mock
public List<Pair<Integer, Long>> getRealOffsets(String brokerList,
String topic,
Map<String,
String> convertedCustomProperties,
-
List<Pair<Integer, Long>> offsetFlags)
+
List<Pair<Integer, Long>> offsetFlags,
+ String
cloudCluster)
throws
LoadException {
+ Assert.assertNull(cloudCluster);
List<Pair<Integer, Long>> pairList = new ArrayList<>();
pairList.add(Pair.of(1, 0L));
return pairList;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
