You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2021/10/15 07:29:13 UTC
[incubator-doris] 03/07: [Optimize][RoutineLoad] Avoid sending
tasks if there is no data to be consumed (#6805)
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-0.15
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
commit 2522df04b8ad9acc84090e45774dcf3f213e1c31
Author: Mingyu Chen <mo...@gmail.com>
AuthorDate: Wed Oct 13 11:39:01 2021 +0800
[Optimize][RoutineLoad] Avoid sending tasks if there is no data to be consumed (#6805)
1 Avoid sending tasks if there is no data to be consumed
By fetching latest offset of partition before sending tasks.(Fix [Optimize] Avoid too many abort task in routine load job #6803 )
2 Add a preCheckNeedSchedule phase in update() of routine load.
To avoid taking write lock of job for long time when getting all kafka partitions from kafka server.
3 Upgrade librdkafka's version to 1.7.0 to fix a bug of "Local: Unknown partition"
See offsetsForTimes fails with 'Local: Unknown partition' edenhill/librdkafka#3295
4 Avoid unnecessary storage migration task if there is no that storage medium on BE.
Fix [Bug] Too many unnecessary storage migration tasks #6804
---
be/src/runtime/routine_load/data_consumer.cpp | 29 +++++-
be/src/runtime/routine_load/data_consumer.h | 3 +
.../routine_load/routine_load_task_executor.cpp | 23 ++++-
.../routine_load/routine_load_task_executor.h | 3 +
be/src/service/internal_service.cpp | 18 +++-
conf/fe.conf | 2 +-
.../java/org/apache/doris/catalog/DiskInfo.java | 9 +-
.../main/java/org/apache/doris/common/Config.java | 2 +-
.../org/apache/doris/common/util/KafkaUtil.java | 70 +++++++++++++-
.../apache/doris/journal/bdbje/BDBJEJournal.java | 2 +-
.../load/routineload/KafkaRoutineLoadJob.java | 106 +++++++++++++++++----
.../doris/load/routineload/KafkaTaskInfo.java | 21 +++-
.../doris/load/routineload/RoutineLoadJob.java | 23 +++--
.../doris/load/routineload/RoutineLoadManager.java | 20 ++--
.../load/routineload/RoutineLoadTaskInfo.java | 4 +-
.../load/routineload/RoutineLoadTaskScheduler.java | 12 ++-
.../org/apache/doris/master/ReportHandler.java | 10 ++
.../main/java/org/apache/doris/system/Backend.java | 4 +
.../java/org/apache/doris/catalog/BackendTest.java | 11 ++-
.../load/routineload/RoutineLoadManagerTest.java | 2 +-
.../routineload/RoutineLoadTaskSchedulerTest.java | 2 +-
gensrc/proto/internal_service.proto | 3 +
thirdparty/vars.sh | 10 +-
23 files changed, 322 insertions(+), 67 deletions(-)
diff --git a/be/src/runtime/routine_load/data_consumer.cpp b/be/src/runtime/routine_load/data_consumer.cpp
index 34cba02..c387b07 100644
--- a/be/src/runtime/routine_load/data_consumer.cpp
+++ b/be/src/runtime/routine_load/data_consumer.cpp
@@ -172,7 +172,6 @@ Status KafkaDataConsumer::assign_topic_partitions(
Status KafkaDataConsumer::group_consume(BlockingQueue<RdKafka::Message*>* queue,
int64_t max_running_time_ms) {
- _last_visit_time = time(nullptr);
int64_t left_time = max_running_time_ms;
LOG(INFO) << "start kafka consumer: " << _id << ", grp: " << _grp_id
<< ", max running time(ms): " << left_time;
@@ -338,7 +337,30 @@ Status KafkaDataConsumer::get_offsets_for_times(const std::vector<PIntegerPair>&
PIntegerPair pair;
pair.set_key(topic_partition->partition());
pair.set_val(topic_partition->offset());
- offsets->push_back(pair);
+ offsets->push_back(std::move(pair));
+ }
+
+ return Status::OK();
+}
+
+// get latest offsets for given partitions
+Status KafkaDataConsumer::get_latest_offsets_for_partitions(const std::vector<int32_t>& partition_ids,
+ std::vector<PIntegerPair>* offsets) {
+ for (int32_t partition_id : partition_ids) {
+ int64_t low = 0;
+ int64_t high = 0;
+ RdKafka::ErrorCode err = _k_consumer->query_watermark_offsets(_topic, partition_id, &low, &high, 5000);
+ if (err != RdKafka::ERR_NO_ERROR) {
+ std::stringstream ss;
+ ss << "failed to get latest offset for partition: " << partition_id << ", err: " << RdKafka::err2str(err);
+ LOG(WARNING) << ss.str();
+ return Status::InternalError(ss.str());
+ }
+
+ PIntegerPair pair;
+ pair.set_key(partition_id);
+ pair.set_val(high);
+ offsets->push_back(std::move(pair));
}
return Status::OK();
@@ -358,6 +380,9 @@ Status KafkaDataConsumer::cancel(StreamLoadContext* ctx) {
Status KafkaDataConsumer::reset() {
std::unique_lock<std::mutex> l(_lock);
_cancelled = false;
+ // reset will be called before this consumer being returned to the pool.
+ // so update _last_visit_time is reasonable.
+ _last_visit_time = time(nullptr);
return Status::OK();
}
diff --git a/be/src/runtime/routine_load/data_consumer.h b/be/src/runtime/routine_load/data_consumer.h
index fcbd017..17b1810 100644
--- a/be/src/runtime/routine_load/data_consumer.h
+++ b/be/src/runtime/routine_load/data_consumer.h
@@ -145,6 +145,9 @@ public:
// get offsets for times
Status get_offsets_for_times(const std::vector<PIntegerPair>& times,
std::vector<PIntegerPair>* offsets);
+ // get latest offsets for partitions
+ Status get_latest_offsets_for_partitions(const std::vector<int32_t>& partition_ids,
+ std::vector<PIntegerPair>* offsets);
private:
std::string _brokers;
diff --git a/be/src/runtime/routine_load/routine_load_task_executor.cpp b/be/src/runtime/routine_load/routine_load_task_executor.cpp
index 0e7ae45..8cc68a6 100644
--- a/be/src/runtime/routine_load/routine_load_task_executor.cpp
+++ b/be/src/runtime/routine_load/routine_load_task_executor.cpp
@@ -39,7 +39,7 @@ RoutineLoadTaskExecutor::RoutineLoadTaskExecutor(ExecEnv* exec_env)
: _exec_env(exec_env),
_thread_pool(config::routine_load_thread_pool_size,
config::routine_load_thread_pool_size),
- _data_consumer_pool(10) {
+ _data_consumer_pool(config::routine_load_thread_pool_size) {
REGISTER_HOOK_METRIC(routine_load_task_count, [this]() {
std::lock_guard<std::mutex> l(_lock);
return _task_map.size();
@@ -126,6 +126,27 @@ Status RoutineLoadTaskExecutor::get_kafka_partition_offsets_for_times(const PKaf
return st;
}
+Status RoutineLoadTaskExecutor::get_kafka_latest_offsets_for_partitions(const PKafkaMetaProxyRequest& request,
+ std::vector<PIntegerPair>* partition_offsets) {
+ CHECK(request.has_kafka_info());
+
+ // This context is meaningless, just for unifing the interface
+ StreamLoadContext ctx(_exec_env);
+ RETURN_IF_ERROR(_prepare_ctx(request, &ctx));
+
+ std::shared_ptr<DataConsumer> consumer;
+ RETURN_IF_ERROR(_data_consumer_pool.get_consumer(&ctx, &consumer));
+
+ Status st = std::static_pointer_cast<KafkaDataConsumer>(consumer)->get_latest_offsets_for_partitions(
+ std::vector<int32_t>(request.partition_id_for_latest_offsets().begin(),
+ request.partition_id_for_latest_offsets().end()),
+ partition_offsets);
+ if (st.ok()) {
+ _data_consumer_pool.return_consumer(consumer);
+ }
+ return st;
+}
+
Status RoutineLoadTaskExecutor::submit_task(const TRoutineLoadTask& task) {
std::unique_lock<std::mutex> l(_lock);
if (_task_map.find(task.id) != _task_map.end()) {
diff --git a/be/src/runtime/routine_load/routine_load_task_executor.h b/be/src/runtime/routine_load/routine_load_task_executor.h
index e5c7939..41f20b1 100644
--- a/be/src/runtime/routine_load/routine_load_task_executor.h
+++ b/be/src/runtime/routine_load/routine_load_task_executor.h
@@ -55,6 +55,9 @@ public:
Status get_kafka_partition_offsets_for_times(const PKafkaMetaProxyRequest& request,
std::vector<PIntegerPair>* partition_offsets);
+ Status get_kafka_latest_offsets_for_partitions(const PKafkaMetaProxyRequest& request,
+ std::vector<PIntegerPair>* partition_offsets);
+
private:
// execute the task
void exec_task(StreamLoadContext* ctx, DataConsumerPool* pool, ExecFinishCallback cb);
diff --git a/be/src/service/internal_service.cpp b/be/src/service/internal_service.cpp
index 89dd7ef..cd98ea0 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -208,7 +208,23 @@ void PInternalServiceImpl<T>::get_info(google::protobuf::RpcController* controll
// 2. get all kafka partition offsets for given topic and timestamp.
if (request->has_kafka_meta_request()) {
const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.offset_times().empty()) {
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st =
+ _exec_env->routine_load_task_executor()->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
+ }
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
// if offset_times() has elements, which means this request is to get offset by timestamp.
std::vector<PIntegerPair> partition_offsets;
Status st =
diff --git a/conf/fe.conf b/conf/fe.conf
index 9298619..723221d 100644
--- a/conf/fe.conf
+++ b/conf/fe.conf
@@ -58,7 +58,7 @@ mysql_service_nio_enabled = true
# log_roll_size_mb = 1024
# sys_log_dir = ${DORIS_HOME}/log
# sys_log_roll_num = 10
-# sys_log_verbose_modules =
+# sys_log_verbose_modules = org.apache.doris
# audit_log_dir = ${DORIS_HOME}/log
# audit_log_modules = slow_query, query
# audit_log_roll_num = 10
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/DiskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/DiskInfo.java
index ee08790..e18b07f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/DiskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/DiskInfo.java
@@ -24,6 +24,8 @@ import org.apache.doris.common.io.Writable;
import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.thrift.TStorageMedium;
+import com.google.gson.annotations.SerializedName;
+
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -31,8 +33,6 @@ import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
-import com.google.gson.annotations.SerializedName;
-
public class DiskInfo implements Writable {
private static final Logger LOG = LogManager.getLogger(DiskInfo.class);
@@ -53,7 +53,6 @@ public class DiskInfo implements Writable {
private long diskAvailableCapacityB;
@SerializedName("state")
private DiskState state;
-
// path hash and storage medium are reported from Backend and no need to persist
private long pathHash = 0;
private TStorageMedium storageMedium;
@@ -129,6 +128,10 @@ public class DiskInfo implements Writable {
return pathHash != 0;
}
+ public boolean isStorageMediumMatch(TStorageMedium storageMedium) {
+ return this.storageMedium == storageMedium;
+ }
+
public TStorageMedium getStorageMedium() {
return storageMedium;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/Config.java b/fe/fe-core/src/main/java/org/apache/doris/common/Config.java
index 2b66c81..d319d3c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/Config.java
@@ -1168,7 +1168,7 @@ public class Config extends ConfigBase {
* not work to avoid OOM.
*/
@ConfField(mutable = true, masterOnly = true)
- public static long metadata_checkpoint_memory_threshold = 60;
+ public static long metadata_checkpoint_memory_threshold = 70;
/**
* If set to true, the checkpoint thread will make the checkpoint regardless of the jvm memory used percent.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/KafkaUtil.java b/fe/fe-core/src/main/java/org/apache/doris/common/util/KafkaUtil.java
index 6691af6..ecbb897 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/KafkaUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/KafkaUtil.java
@@ -37,6 +37,7 @@ import org.apache.logging.log4j.Logger;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -144,7 +145,7 @@ public class KafkaUtil {
InternalService.PProxyResult result = future.get(5, TimeUnit.SECONDS);
TStatusCode code = TStatusCode.findByValue(result.getStatus().getStatusCode());
if (code != TStatusCode.OK) {
- throw new UserException("failed to get kafka partition info: " + result.getStatus().getErrorMsgsList());
+ throw new UserException("failed to get offsets for times: " + result.getStatus().getErrorMsgsList());
} else {
List<InternalService.PIntegerPair> pairs = result.getPartitionOffsets().getOffsetTimesList();
List<Pair<Integer, Long>> partitionOffsets = Lists.newArrayList();
@@ -166,6 +167,73 @@ 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 {
+ BackendService.Client client = null;
+ TNetworkAddress address = null;
+ LOG.debug("begin to get latest offsets for partitions {} in topic: {}, task {}, job {}",
+ partitionIds, topic, taskId, jobId);
+ boolean ok = false;
+ try {
+ List<Long> backendIds = Catalog.getCurrentSystemInfo().getBackendIds(true);
+ if (backendIds.isEmpty()) {
+ throw new LoadException("Failed to get latest offsets. No alive backends");
+ }
+ Collections.shuffle(backendIds);
+ Backend be = Catalog.getCurrentSystemInfo().getBackend(backendIds.get(0));
+ address = new TNetworkAddress(be.getHost(), be.getBrpcPort());
+
+ // create request
+ InternalService.PKafkaMetaProxyRequest.Builder metaRequestBuilder =
+ InternalService.PKafkaMetaProxyRequest.newBuilder()
+ .setKafkaInfo(InternalService.PKafkaLoadInfo.newBuilder()
+ .setBrokers(brokerList)
+ .setTopic(topic)
+ .addAllProperties(
+ convertedCustomProperties.entrySet().stream().map(
+ e -> InternalService.PStringPair.newBuilder()
+ .setKey(e.getKey())
+ .setVal(e.getValue())
+ .build()
+ ).collect(Collectors.toList())
+ )
+ );
+ for (Integer partitionId : partitionIds) {
+ metaRequestBuilder.addPartitionIdForLatestOffsets(partitionId);
+ }
+ InternalService.PProxyRequest request = InternalService.PProxyRequest.newBuilder().setKafkaMetaRequest(
+ metaRequestBuilder).build();
+
+ // get info
+ Future<InternalService.PProxyResult> future = BackendServiceProxy.getInstance().getInfo(address, request);
+ InternalService.PProxyResult result = future.get(5, TimeUnit.SECONDS);
+ TStatusCode code = TStatusCode.findByValue(result.getStatus().getStatusCode());
+ if (code != TStatusCode.OK) {
+ throw new UserException("failed to get latest offsets: " + result.getStatus().getErrorMsgsList());
+ } else {
+ List<InternalService.PIntegerPair> pairs = result.getPartitionOffsets().getOffsetTimesList();
+ List<Pair<Integer, Long>> partitionOffsets = Lists.newArrayList();
+ for (InternalService.PIntegerPair pair : pairs) {
+ partitionOffsets.add(Pair.create(pair.getKey(), pair.getVal()));
+ }
+ LOG.debug("finish to get latest offsets for partitions {} in topic: {}, task {}, job {}",
+ partitionOffsets, topic, taskId, jobId);
+ return partitionOffsets;
+ }
+ } catch (Exception e) {
+ LOG.warn("failed to get latest offsets.", e);
+ throw new LoadException(
+ "Failed to get latest offsets of kafka topic: " + topic + ". error: " + e.getMessage());
+ } finally {
+ if (ok) {
+ ClientPool.backendPool.returnObject(address, client);
+ } else {
+ ClientPool.backendPool.invalidateObject(address, client);
+ }
+ }
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/journal/bdbje/BDBJEJournal.java b/fe/fe-core/src/main/java/org/apache/doris/journal/bdbje/BDBJEJournal.java
index 553aa22..d1c29fe 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/journal/bdbje/BDBJEJournal.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/journal/bdbje/BDBJEJournal.java
@@ -446,4 +446,4 @@ public class BDBJEJournal implements Journal {
return dbNames;
}
-}
\ No newline at end of file
+}
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 7a3b87b..1be8030 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
@@ -91,6 +91,14 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
private Map<String, String> customProperties = Maps.newHashMap();
private Map<String, String> convertedCustomProperties = Maps.newHashMap();
+ // The latest offset of each partition fetched from kafka server.
+ // Will be updated periodically by calling hasMoreDataToConsume()
+ private Map<Integer, Long> cachedPartitionWithLatestOffsets = Maps.newConcurrentMap();
+
+ // The kafka partition fetch from kafka server.
+ // Will be updated periodically by calling updateKafkaPartitions();
+ private List<Integer> newCurrentKafkaPartition = Lists.newArrayList();
+
public KafkaRoutineLoadJob() {
// for serialization, id is dummy
super(-1, LoadDataSourceType.KAFKA);
@@ -281,6 +289,34 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
updateNewPartitionProgress();
}
+ @Override
+ protected void preCheckNeedSchedule() throws UserException {
+ // If user does not specify kafka partition,
+ // We will fetch partition from kafka server periodically
+ if (this.state == JobState.RUNNING || this.state == JobState.NEED_SCHEDULE) {
+ if (customKafkaPartitions == null && !customKafkaPartitions.isEmpty()) {
+ return;
+ }
+ updateKafkaPartitions();
+ }
+ }
+
+ private void updateKafkaPartitions() throws UserException {
+ try {
+ this.newCurrentKafkaPartition = getAllKafkaPartitions();
+ } catch (Exception e) {
+ LOG.warn(new LogBuilder(LogKey.ROUTINE_LOAD_JOB, id)
+ .add("error_msg", "Job failed to fetch all current partition with error " + e.getMessage())
+ .build(), e);
+ if (this.state == JobState.NEED_SCHEDULE) {
+ unprotectUpdateState(JobState.PAUSED,
+ new ErrorReason(InternalErrorCode.PARTITIONS_ERR,
+ "Job failed to fetch all current partition with error " + e.getMessage()),
+ false /* not replay */);
+ }
+ }
+ }
+
// if customKafkaPartition is not null, then return false immediately
// else if kafka partitions of topic has been changed, return true.
// else return false
@@ -294,24 +330,11 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
currentKafkaPartitions = customKafkaPartitions;
return false;
} else {
- List<Integer> newCurrentKafkaPartition;
- try {
- newCurrentKafkaPartition = getAllKafkaPartitions();
- } catch (Exception e) {
- LOG.warn(new LogBuilder(LogKey.ROUTINE_LOAD_JOB, id)
- .add("error_msg", "Job failed to fetch all current partition with error " + e.getMessage())
- .build(), e);
- if (this.state == JobState.NEED_SCHEDULE) {
- unprotectUpdateState(JobState.PAUSED,
- new ErrorReason(InternalErrorCode.PARTITIONS_ERR,
- "Job failed to fetch all current partition with error " + e.getMessage()),
- false /* not replay */);
- }
- return false;
- }
- if (currentKafkaPartitions.containsAll(newCurrentKafkaPartition)) {
- if (currentKafkaPartitions.size() > newCurrentKafkaPartition.size()) {
- currentKafkaPartitions = newCurrentKafkaPartition;
+ // the newCurrentKafkaPartition should be already updated in preCheckNeedScheduler()
+ Preconditions.checkNotNull(this.newCurrentKafkaPartition);
+ if (currentKafkaPartitions.containsAll(this.newCurrentKafkaPartition)) {
+ if (currentKafkaPartitions.size() > this.newCurrentKafkaPartition.size()) {
+ currentKafkaPartitions = this.newCurrentKafkaPartition;
if (LOG.isDebugEnabled()) {
LOG.debug(new LogBuilder(LogKey.ROUTINE_LOAD_JOB, id)
.add("current_kafka_partitions", Joiner.on(",").join(currentKafkaPartitions))
@@ -323,7 +346,7 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
return false;
}
} else {
- currentKafkaPartitions = newCurrentKafkaPartition;
+ currentKafkaPartitions = this.newCurrentKafkaPartition;
if (LOG.isDebugEnabled()) {
LOG.debug(new LogBuilder(LogKey.ROUTINE_LOAD_JOB, id)
.add("current_kafka_partitions", Joiner.on(",").join(currentKafkaPartitions))
@@ -624,4 +647,49 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
LOG.error("failed to replay modify kafka routine load job: {}", id, e);
}
}
+
+ // check if given partitions has more data to consume.
+ // 'partitionIdToOffset' to the offset to be consumed.
+ public boolean hasMoreDataToConsume(UUID taskId, Map<Integer, Long> partitionIdToOffset) {
+ for (Map.Entry<Integer, Long> entry : partitionIdToOffset.entrySet()) {
+ if (cachedPartitionWithLatestOffsets.containsKey(entry.getKey())
+ && entry.getValue() < cachedPartitionWithLatestOffsets.get(entry.getKey())) {
+ // "entry.getValue()" is the offset to be consumed.
+ // "cachedPartitionWithLatestOffsets.get(entry.getKey())" is the "next" offset of this partition.
+ // (because librdkafa's query_watermark_offsets() will return the next offset.
+ // For example, there 4 msg in partition with offset 0,1,2,3,
+ // query_watermark_offsets() will return 4.)
+ LOG.debug("has more data to consume. offsets to be consumed: {}, latest offsets: {}, task {}, job {}",
+ partitionIdToOffset, cachedPartitionWithLatestOffsets, taskId, id);
+ return true;
+ }
+ }
+
+ try {
+ // all offsets to be consumed are newer than offsets in cachedPartitionWithLatestOffsets,
+ // maybe the cached offset is out-of-date, fetch from kafka server again
+ List<Pair<Integer, Long>> tmp = KafkaUtil.getLatestOffsets(id, taskId, getBrokerList(),
+ getTopic(), getConvertedCustomProperties(), Lists.newArrayList(partitionIdToOffset.keySet()));
+ for (Pair<Integer, Long> pair : tmp) {
+ cachedPartitionWithLatestOffsets.put(pair.first, pair.second);
+ }
+ } catch (Exception e) {
+ LOG.warn("failed to get latest partition offset. {}", e.getMessage(), e);
+ return false;
+ }
+
+ // check again
+ for (Map.Entry<Integer, Long> entry : partitionIdToOffset.entrySet()) {
+ if (cachedPartitionWithLatestOffsets.containsKey(entry.getKey())
+ && entry.getValue() < cachedPartitionWithLatestOffsets.get(entry.getKey())) {
+ LOG.debug("has more data to consume. offsets to be consumed: {}, latest offsets: {}, task {}, job {}",
+ partitionIdToOffset, cachedPartitionWithLatestOffsets, taskId, id);
+ return true;
+ }
+ }
+
+ LOG.debug("no more data to consume. offsets to be consumed: {}, latest offsets: {}, task {}, job {}",
+ partitionIdToOffset, cachedPartitionWithLatestOffsets, taskId, id);
+ return false;
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
index f4d4d6e..c7dfd60 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
@@ -20,6 +20,7 @@ package org.apache.doris.load.routineload;
import org.apache.doris.catalog.Catalog;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Table;
+import org.apache.doris.common.Pair;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.thrift.TExecPlanFragmentParams;
@@ -31,20 +32,28 @@ import org.apache.doris.thrift.TRoutineLoadTask;
import org.apache.doris.thrift.TUniqueId;
import com.google.common.base.Joiner;
+import com.google.common.collect.Lists;
import com.google.gson.Gson;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
public class KafkaTaskInfo extends RoutineLoadTaskInfo {
+ private static final Logger LOG = LogManager.getLogger(KafkaTaskInfo.class);
private RoutineLoadManager routineLoadManager = Catalog.getCurrentCatalog().getRoutineLoadManager();
- // <partitionId, beginOffsetOfPartitionId>
+ // <partitionId, offset to be consumed>
private Map<Integer, Long> partitionIdToOffset;
+ // Last fetched and cached latest partition offsets.
+ private List<Pair<Integer, Long>> cachedPartitionWithLatestOffsets = Lists.newArrayList();
+
public KafkaTaskInfo(UUID id, long jobId, String clusterName, long timeoutMs, Map<Integer, Long> partitionIdToOffset) {
super(id, jobId, clusterName, timeoutMs);
this.partitionIdToOffset = partitionIdToOffset;
@@ -79,8 +88,8 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
tRoutineLoadTask.setLabel(label);
tRoutineLoadTask.setAuthCode(routineLoadJob.getAuthCode());
TKafkaLoadInfo tKafkaLoadInfo = new TKafkaLoadInfo();
- tKafkaLoadInfo.setTopic((routineLoadJob).getTopic());
- tKafkaLoadInfo.setBrokers((routineLoadJob).getBrokerList());
+ tKafkaLoadInfo.setTopic(routineLoadJob.getTopic());
+ tKafkaLoadInfo.setBrokers(routineLoadJob.getBrokerList());
tKafkaLoadInfo.setPartitionBeginOffset(partitionIdToOffset);
tKafkaLoadInfo.setProperties(routineLoadJob.getConvertedCustomProperties());
tRoutineLoadTask.setKafkaLoadInfo(tKafkaLoadInfo);
@@ -103,6 +112,12 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
return gson.toJson(partitionIdToOffset);
}
+ @Override
+ boolean hasMoreDataToConsume() {
+ KafkaRoutineLoadJob routineLoadJob = (KafkaRoutineLoadJob) routineLoadManager.getJob(jobId);
+ return routineLoadJob.hasMoreDataToConsume(id, partitionIdToOffset);
+ }
+
private TExecPlanFragmentParams rePlan(RoutineLoadJob routineLoadJob) throws UserException {
TUniqueId loadId = new TUniqueId(id.getMostSignificantBits(), id.getLeastSignificantBits());
// plan for each task, in case table has change(rollup or schema change)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
index 734c8b3..b566029 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
@@ -637,11 +637,7 @@ public abstract class RoutineLoadJob extends AbstractTxnStateChangeCallback impl
for (RoutineLoadTaskInfo routineLoadTaskInfo : routineLoadTaskInfoList) {
if (routineLoadTaskInfo.getBeId() != -1L) {
long beId = routineLoadTaskInfo.getBeId();
- if (beIdConcurrentTasksNum.containsKey(beId)) {
- beIdConcurrentTasksNum.put(beId, beIdConcurrentTasksNum.get(beId) + 1);
- } else {
- beIdConcurrentTasksNum.put(beId, 1);
- }
+ beIdConcurrentTasksNum.put(beId, beIdConcurrentTasksNum.getOrDefault(beId, 0) + 1);
}
}
return beIdConcurrentTasksNum;
@@ -1205,7 +1201,7 @@ public abstract class RoutineLoadJob extends AbstractTxnStateChangeCallback impl
try {
if (!state.isFinalState()) {
unprotectUpdateState(JobState.CANCELLED,
- new ErrorReason(InternalErrorCode.TABLE_ERR, "table not exist"), false /* not replay */);
+ new ErrorReason(InternalErrorCode.TABLE_ERR, "table does not exist"), false /* not replay */);
}
return;
} finally {
@@ -1213,13 +1209,14 @@ public abstract class RoutineLoadJob extends AbstractTxnStateChangeCallback impl
}
}
- // check if partition has been changed
+ preCheckNeedSchedule();
+
writeLock();
try {
if (unprotectNeedReschedule()) {
LOG.info(new LogBuilder(LogKey.ROUTINE_LOAD_JOB, id)
- .add("msg", "Job need to be rescheduled")
- .build());
+ .add("msg", "Job need to be rescheduled")
+ .build());
unprotectUpdateProgress();
executeNeedSchedule();
}
@@ -1228,6 +1225,14 @@ public abstract class RoutineLoadJob extends AbstractTxnStateChangeCallback impl
}
}
+ // Call this before calling unprotectUpdateProgress().
+ // Because unprotectUpdateProgress() is protected by writelock.
+ // So if there are time-consuming operations, they should be done in this method.
+ // (Such as getAllKafkaPartitions() in KafkaRoutineLoad)
+ protected void preCheckNeedSchedule() throws UserException {
+
+ }
+
protected void unprotectUpdateProgress() throws UserException {
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
index e6a5263..4d283e5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadManager.java
@@ -41,14 +41,14 @@ import org.apache.doris.persist.AlterRoutineLoadJobOperationLog;
import org.apache.doris.persist.RoutineLoadOperation;
import org.apache.doris.qe.ConnectContext;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
@@ -58,7 +58,6 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.UUID;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
@@ -355,7 +354,6 @@ public class RoutineLoadManager implements Writable {
readLock();
try {
int result = 0;
- updateBeIdToMaxConcurrentTasks();
Map<Long, Integer> beIdToConcurrentTasks = getBeCurrentTasksNumMap();
for (Map.Entry<Long, Integer> entry : beIdToMaxConcurrentTasks.entrySet()) {
if (beIdToConcurrentTasks.containsKey(entry.getKey())) {
@@ -439,7 +437,6 @@ public class RoutineLoadManager implements Writable {
}
// 2. The given BE id does not have available slots, find a BE with min tasks
- updateBeIdToMaxConcurrentTasks();
int idleTaskNum = 0;
long resultBeId = -1L;
int maxIdleSlotNum = 0;
@@ -558,13 +555,12 @@ public class RoutineLoadManager implements Writable {
}
}
- public boolean checkTaskInJob(UUID taskId) {
- for (RoutineLoadJob routineLoadJob : idToRoutineLoadJob.values()) {
- if (routineLoadJob.containsTask(taskId)) {
- return true;
- }
+ public boolean checkTaskInJob(RoutineLoadTaskInfo task) {
+ RoutineLoadJob routineLoadJob = idToRoutineLoadJob.get(task.getJobId());
+ if (routineLoadJob == null) {
+ return false;
}
- return false;
+ return routineLoadJob.containsTask(task.getId());
}
public List<RoutineLoadJob> getRoutineLoadJobByState(Set<RoutineLoadJob.JobState> desiredStates) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskInfo.java
index 71bdb99..b535b94 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskInfo.java
@@ -207,7 +207,9 @@ public abstract class RoutineLoadTaskInfo {
}
abstract String getTaskDataSourceProperties();
-
+
+ abstract boolean hasMoreDataToConsume();
+
@Override
public boolean equals(Object obj) {
if (obj instanceof RoutineLoadTaskInfo) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
index e204155..b83fa13 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
@@ -87,8 +87,10 @@ public class RoutineLoadTaskScheduler extends MasterDaemon {
}
private void process() throws UserException, InterruptedException {
+ // update the max slot num of each backend periodically
updateBackendSlotIfNecessary();
+ long start = System.currentTimeMillis();
// if size of queue is zero, tasks will be submitted by batch
int idleSlotNum = routineLoadManager.getClusterIdleSlotNum();
// scheduler will be blocked when there is no slot for task in cluster
@@ -114,8 +116,9 @@ public class RoutineLoadTaskScheduler extends MasterDaemon {
private void scheduleOneTask(RoutineLoadTaskInfo routineLoadTaskInfo) throws Exception {
routineLoadTaskInfo.setLastScheduledTime(System.currentTimeMillis());
+ LOG.debug("schedule routine load task info {} for job {}", routineLoadTaskInfo.id, routineLoadTaskInfo.getJobId());
// check if task has been abandoned
- if (!routineLoadManager.checkTaskInJob(routineLoadTaskInfo.getId())) {
+ if (!routineLoadManager.checkTaskInJob(routineLoadTaskInfo)) {
// task has been abandoned while renew task has been added in queue
// or database has been deleted
LOG.warn(new LogBuilder(LogKey.ROUTINE_LOAD_TASK, routineLoadTaskInfo.getId())
@@ -124,6 +127,12 @@ public class RoutineLoadTaskScheduler extends MasterDaemon {
return;
}
+ // check if topic has more data to consume
+ if (!routineLoadTaskInfo.hasMoreDataToConsume()) {
+ needScheduleTasksQueue.put(routineLoadTaskInfo);
+ return;
+ }
+
// allocate BE slot for this task.
// this should be done before txn begin, or the txn may be begun successfully but failed to be allocated.
try {
@@ -294,3 +303,4 @@ public class RoutineLoadTaskScheduler extends MasterDaemon {
return true;
}
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java b/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java
index df0e599..8ee321e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java
@@ -759,9 +759,19 @@ public class ReportHandler extends Daemon {
private static void handleMigration(ListMultimap<TStorageMedium, Long> tabletMetaMigrationMap,
long backendId) {
TabletInvertedIndex invertedIndex = Catalog.getCurrentInvertedIndex();
+ SystemInfoService infoService = Catalog.getCurrentSystemInfo();
+ Backend be = infoService.getBackend(backendId);
+ if (be == null) {
+ return;
+ }
AgentBatchTask batchTask = new AgentBatchTask();
for (TStorageMedium storageMedium : tabletMetaMigrationMap.keySet()) {
List<Long> tabletIds = tabletMetaMigrationMap.get(storageMedium);
+ if (!be.hasSpecifiedStorageMedium(storageMedium)) {
+ LOG.warn("no specified storage medium {} on backend {}, skip storage migration." +
+ " sample tablet id: {}", storageMedium, backendId, tabletIds.isEmpty() ? "-1" : tabletIds.get(0));
+ continue;
+ }
List<TabletMeta> tabletMetaList = invertedIndex.getTabletMetaList(tabletIds);
for (int i = 0; i < tabletMetaList.size(); i++) {
long tabletId = tabletIds.get(i);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java b/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
index 0c5d66d..03fd69a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
@@ -330,6 +330,10 @@ public class Backend implements Writable {
return disksRef.values().stream().allMatch(DiskInfo::hasPathHash);
}
+ public boolean hasSpecifiedStorageMedium(TStorageMedium storageMedium) {
+ return disksRef.values().stream().anyMatch(d -> d.isStorageMediumMatch(storageMedium));
+ }
+
public long getTotalCapacityB() {
ImmutableMap<String, DiskInfo> disks = disksRef;
long totalCapacityB = 0L;
diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/BackendTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/BackendTest.java
index 34c4fb4..c4f6913 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/BackendTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/BackendTest.java
@@ -21,9 +21,14 @@ import org.apache.doris.analysis.AccessTestUtil;
import org.apache.doris.common.FeConstants;
import org.apache.doris.system.Backend;
import org.apache.doris.thrift.TDisk;
+import org.apache.doris.thrift.TStorageMedium;
import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
@@ -34,10 +39,6 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
public class BackendTest {
private Backend backend;
private long backendId = 9999;
@@ -107,6 +108,8 @@ public class BackendTest {
backend.updateDisks(diskInfos);
Assert.assertEquals(disk2.getDiskTotalCapacity(), backend.getTotalCapacityB());
Assert.assertEquals(disk2.getDiskAvailableCapacity() + 1, backend.getAvailableCapacityB());
+ Assert.assertFalse(backend.hasSpecifiedStorageMedium(TStorageMedium.SSD));
+ Assert.assertTrue(backend.hasSpecifiedStorageMedium(TStorageMedium.HDD));
}
@Test
diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadManagerTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadManagerTest.java
index 0512e3e..95f4889 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadManagerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadManagerTest.java
@@ -417,7 +417,7 @@ public class RoutineLoadManagerTest {
};
Deencapsulation.setField(routineLoadManager, "idToRoutineLoadJob", idToRoutineLoadJob);
-
+ routineLoadManager.updateBeIdToMaxConcurrentTasks();
Assert.assertEquals(Config.max_routine_load_task_num_per_be * 2 - 1,
routineLoadManager.getClusterIdleSlotNum());
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadTaskSchedulerTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadTaskSchedulerTest.java
index 133dcb8..8cb98d6 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadTaskSchedulerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadTaskSchedulerTest.java
@@ -95,7 +95,7 @@ public class RoutineLoadTaskSchedulerTest {
routineLoadManager.getClusterIdleSlotNum();
minTimes = 0;
result = 1;
- routineLoadManager.checkTaskInJob((UUID) any);
+ routineLoadManager.checkTaskInJob((RoutineLoadTaskInfo) any);
minTimes = 0;
result = true;
diff --git a/gensrc/proto/internal_service.proto b/gensrc/proto/internal_service.proto
index 016f4d1..5ecbb69 100644
--- a/gensrc/proto/internal_service.proto
+++ b/gensrc/proto/internal_service.proto
@@ -239,7 +239,10 @@ message PKafkaLoadInfo {
message PKafkaMetaProxyRequest {
optional PKafkaLoadInfo kafka_info = 1;
+ // optional for getting offsets for times
repeated PIntegerPair offset_times = 3;
+ // optional for getting latest offsets of partitons
+ repeated int32 partition_id_for_latest_offsets = 4;
};
message PProxyRequest {
diff --git a/thirdparty/vars.sh b/thirdparty/vars.sh
index f202d14..5aa6efa 100644
--- a/thirdparty/vars.sh
+++ b/thirdparty/vars.sh
@@ -202,11 +202,11 @@ ROCKSDB_NAME=rocksdb-5.14.2.tar.gz
ROCKSDB_SOURCE=rocksdb-5.14.2
ROCKSDB_MD5SUM="b72720ea3b1e9ca9e4ed0febfef65b14"
-# librdkafka-1.6.1
-LIBRDKAFKA_DOWNLOAD="https://github.com/edenhill/librdkafka/archive/v1.6.1-RC3.tar.gz"
-LIBRDKAFKA_NAME=librdkafka-1.6.1-RC3.tar.gz
-LIBRDKAFKA_SOURCE=librdkafka-1.6.1-RC3
-LIBRDKAFKA_MD5SUM="11b02507db989a0fb5220fcc4df8f79a"
+# librdkafka-1.7.0
+LIBRDKAFKA_DOWNLOAD="https://github.com/edenhill/librdkafka/archive/refs/tags/v1.7.0.tar.gz"
+LIBRDKAFKA_NAME=v1.7.0.tar.gz
+LIBRDKAFKA_SOURCE=librdkafka-1.7.0
+LIBRDKAFKA_MD5SUM="fe3c45deb182bd9c644b6bc6375bffc3"
# zstd
ZSTD_DOWNLOAD="https://github.com/facebook/zstd/archive/v1.3.7.tar.gz"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org