This is an automated email from the ASF dual-hosted git repository.
liaoxin01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a34de28b2c7 [fix](job) keep Kafka metadata requests in compute group
(#66089)
a34de28b2c7 is described below
commit a34de28b2c799df335bd6a98b3e83f6e92ab5431
Author: hui lai <[email protected]>
AuthorDate: Sat Aug 8 17:18:13 2026 +0800
[fix](job) keep Kafka metadata requests in compute group (#66089)
### Problem Summary:
Kafka routine load metadata requests selected backend nodes from all
compute groups, so partition and offset requests could be sent outside
the compute group bound to the job. Missing or stale compute group names
also produced unclear backend-selection failures.
### Solution:
Pass the routine load job persisted compute group to KafkaUtil. In cloud
mode, restrict both normal selection and blacklist fallback to backend
nodes in that compute group, reject a missing compute group before
lookup, and include the group name in terminal failures. Preserve the
existing global selection behavior for non-cloud deployments.
### Tests:
- Add KafkaUtilTest coverage for compute-group-scoped backend selection
and blacklist fallback, missing compute groups, scoped failure messages,
and non-cloud selection.
- Update routine load tests for compute group propagation and the
KafkaUtil API change.
---
.../apache/doris/datasource/kafka/KafkaUtil.java | 121 +++++++++++------
.../routineload/kafka/KafkaRoutineLoadJob.java | 33 +++--
.../doris/datasource/kafka/KafkaUtilTest.java | 150 +++++++++++++++++++++
.../load/routineload/KafkaRoutineLoadJobTest.java | 43 +++---
.../doris/load/routineload/RoutineLoadJobTest.java | 5 +-
5 files changed, 286 insertions(+), 66 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/kafka/KafkaRoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
index 502202aa8cc..88502144035 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
@@ -552,7 +552,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)
@@ -652,13 +657,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)
@@ -690,15 +696,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) {
@@ -796,11 +805,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);
}
@@ -975,8 +987,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.
@@ -1039,8 +1053,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 77570bb0309..7f0c8588372 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
@@ -180,21 +180,29 @@ 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));
+ 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));
- try (MockedStatic<KafkaUtil> kafkaUtilStatic =
Mockito.mockStatic(KafkaUtil.class)) {
- kafkaUtilStatic.when(() ->
KafkaUtil.getLatestOffsets(Mockito.eq(1L), Mockito.any(UUID.class),
- Mockito.eq("127.0.0.1:9020"), Mockito.eq("topic1"),
Mockito.anyMap(), Mockito.anyList()))
- .thenReturn(Lists.newArrayList(Pair.of(1, 15L), Pair.of(2,
30L)));
+ try (MockedStatic<KafkaUtil> kafkaUtilStatic =
Mockito.mockStatic(KafkaUtil.class)) {
+ kafkaUtilStatic.when(() ->
KafkaUtil.getLatestOffsets(Mockito.eq(1L), Mockito.any(UUID.class),
+ Mockito.eq("127.0.0.1:9020"), Mockito.eq("topic1"),
Mockito.anyMap(), Mockito.anyList(),
+ Mockito.eq("routine-load-compute-group")))
+ .thenReturn(Lists.newArrayList(Pair.of(1, 15L),
Pair.of(2, 30L)));
- routineLoadJob.updateLag();
+ routineLoadJob.updateLag();
- Assert.assertEquals(15L, routineLoadJob.totalLag().longValue());
+ Assert.assertEquals(15L,
routineLoadJob.totalLag().longValue());
+ }
+ } finally {
+ Config.cloud_unique_id = originalCloudUniqueId;
}
}
@@ -223,7 +231,8 @@ public class KafkaRoutineLoadJobTest {
Mockito.<Map<String, String>>argThat(properties ->
"SASL_PLAINTEXT".equals(properties.get("security.protocol"))
&&
"PLAIN".equals(properties.get("sasl.mechanism"))),
- Mockito.argThat(partitions -> partitions.size() == 1 &&
partitions.contains(1))))
+ Mockito.argThat(partitions -> partitions.size() == 1 &&
partitions.contains(1)),
+ Mockito.nullable(String.class)))
.thenReturn(Lists.newArrayList(Pair.of(1, 15L)));
routineLoadJob.updateLag();
@@ -234,7 +243,8 @@ public class KafkaRoutineLoadJobTest {
Mockito.<Map<String, String>>argThat(properties ->
"SASL_PLAINTEXT".equals(properties.get("security.protocol"))
&&
"PLAIN".equals(properties.get("sasl.mechanism"))),
- Mockito.argThat(partitions -> partitions.size() == 1 &&
partitions.contains(1))));
+ Mockito.argThat(partitions -> partitions.size() == 1 &&
partitions.contains(1)),
+ Mockito.nullable(String.class)));
}
}
@@ -613,11 +623,12 @@ public class KafkaRoutineLoadJobTest {
.getPartition(Mockito.anyString(), Mockito.anyBoolean());
kafkaUtilStatic.when(() -> KafkaUtil.getAllKafkaPartitions(
- Mockito.anyString(), Mockito.anyString(),
Mockito.anyMap()))
+ Mockito.anyString(), Mockito.anyString(),
Mockito.anyMap(), Mockito.nullable(String.class)))
.thenReturn(Lists.newArrayList(1, 2, 3));
kafkaUtilStatic.when(() -> KafkaUtil.getRealOffsets(
- Mockito.anyString(), Mockito.anyString(),
Mockito.anyMap(), Mockito.anyList()))
+ Mockito.anyString(), Mockito.anyString(),
Mockito.anyMap(), Mockito.anyList(),
+ Mockito.nullable(String.class)))
.thenAnswer(invocation -> {
List<Pair<Integer, Long>> pairList = new ArrayList<>();
pairList.add(Pair.of(1, 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 26351597543..d2724063abf 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
@@ -263,11 +263,12 @@ public class RoutineLoadJobTest {
Mockito.doReturn(table).when(database).getTableNullable(Mockito.anyLong());
kafkaUtilStatic.when(() -> KafkaUtil.getAllKafkaPartitions(
- Mockito.anyString(), Mockito.anyString(),
Mockito.anyMap()))
+ Mockito.anyString(), Mockito.anyString(),
Mockito.anyMap(), Mockito.nullable(String.class)))
.thenReturn(Lists.newArrayList(1, 2, 3));
kafkaUtilStatic.when(() -> KafkaUtil.getRealOffsets(
- Mockito.anyString(), Mockito.anyString(),
Mockito.anyMap(), Mockito.anyList()))
+ Mockito.anyString(), Mockito.anyString(),
Mockito.anyMap(), Mockito.anyList(),
+ Mockito.nullable(String.class)))
.thenAnswer(inv -> {
List<Pair<Integer, Long>> pairList = new ArrayList<>();
pairList.add(Pair.of(1, 0L));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
