github-actions[bot] commented on code in PR #68357:
URL: https://github.com/apache/doris/pull/68357#discussion_r4089585135
##########
gensrc/proto/internal_service.proto:
##########
@@ -471,17 +471,21 @@ message PKinesisLoadInfo {
message PKinesisMetaProxyRequest {
optional PKinesisLoadInfo kinesis_info = 1;
+ // Resolve initial LATEST positions before data tasks are created.
+ repeated string shard_ids_for_latest_sequences = 2;
};
message PShardInfo {
required string shard_id = 1;
optional string parent_shard_id = 2;
optional string adjacent_parent_shard_id = 3;
+ optional bool closed = 4;
};
message PKinesisMetaProxyResult {
- repeated string shard_ids = 1; // Deprecated, use shard_infos instead
- repeated PShardInfo shard_infos = 2;
+ repeated PShardInfo shard_infos = 1;
Review Comment:
[P1] Keep the existing field numbers for the supported old-FE/new-BE upgrade
phase. Before this change tag 1 was `repeated string shard_ids` and tag 2 was
`repeated PShardInfo shard_infos`; because the replacement payloads are also
length-delimited, the old FE decodes serialized `PShardInfo` bytes as shard IDs
instead of skipping them. Please preserve `shard_ids = 1`, keep the full
retained open/closed `shard_infos = 2`, and put `shard_latest_sequences` on a
new tag. Populate legacy field 1 with OPEN shard IDs only, matching the
pre-change BE behavior; populating it from all retained descriptors would make
an old FE revive completed parents. Add a mixed-version serialization fixture
for this direction.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java:
##########
@@ -128,8 +133,19 @@ public class KinesisRoutineLoadJob extends RoutineLoadJob {
// Will be updated periodically by calling hasMoreDataToConsume()
private Map<String, Long> cachedShardWithMillsBehindLatest =
Maps.newConcurrentMap();
- // newly discovered shards from Kinesis.
- private List<String> newCurrentKinesisShards = Lists.newArrayList();
+ // Compatibility views for SHOW/old unit fixtures. Topology remains the
only source of truth.
Review Comment:
[P1] Please migrate the old persisted shard boundary before making these
lists transient. Existing images contain `opks`/`clks` but no `topo`; Gson now
ignores those keys, constructs an empty topology, and `gsonPostProcess()` only
reconciles positions into nodes that already exist. The first post-upgrade
ListShards result is consequently treated as a fresh initial snapshot. In
particular, an already completed parent was removed from old progress and
`clks` but can still be retained by Kinesis, so this code recreates and
schedules it from the default position, duplicating data. Keep legacy holders
for migration and build the initial topology/boundary from them before
accepting the full retained-shard snapshot; add a pre-change image fixture
covering a completed retained parent.
##########
be/src/load/routine_load/routine_load_task_executor.cpp:
##########
@@ -176,13 +195,117 @@ Status
RoutineLoadTaskExecutor::get_kinesis_shard_meta(const PKinesisMetaProxyRe
std::shared_ptr<DataConsumer> consumer;
RETURN_IF_ERROR(_data_consumer_pool.get_consumer(ctx, &consumer));
- Status st =
std::static_pointer_cast<KinesisDataConsumer>(consumer)->get_shard_list(shard_ids);
+ Status st =
+
std::static_pointer_cast<KinesisDataConsumer>(consumer)->get_shard_list(shard_infos);
if (st.ok()) {
_data_consumer_pool.return_consumer(consumer);
}
return st;
}
+// All workers of one RPC share a deadline and publish results only after all
workers exit.
+struct KinesisLatestSequenceBatch {
+ PKinesisMetaProxyRequest request;
+ int64_t deadline_ms;
+ std::function<bool()> is_cancelled;
+ RoutineLoadTaskExecutor::KinesisScanCallback on_finish;
+ std::atomic<int> next_shard {0};
+ std::atomic<int> remaining_workers;
+ std::mutex mutex;
+ Status status;
+ std::map<std::string, std::string> sequences;
+
+ KinesisLatestSequenceBatch(PKinesisMetaProxyRequest req, int64_t
timeout_ms,
+ std::function<bool()> cancelled,
+ RoutineLoadTaskExecutor::KinesisScanCallback
finish, int workers)
+ : request(std::move(req)),
+ deadline_ms(timeout_ms == -1 ? -1 : MonotonicMillis() +
timeout_ms),
+ is_cancelled(std::move(cancelled)),
+ on_finish(std::move(finish)),
+ remaining_workers(workers) {}
+
+ Status check_status() {
+ std::lock_guard<std::mutex> lock(mutex);
+ RETURN_IF_ERROR(status);
+ if (is_cancelled()) {
+ return Status::Cancelled("Kinesis latest sequence scan cancelled");
+ }
+ if (deadline_ms != -1 && MonotonicMillis() >= deadline_ms) {
+ return Status::TimedOut("Kinesis latest sequence scan exceeded its
total timeout");
+ }
+ return Status::OK();
+ }
+
+ void finish_worker(const Status& worker_status) {
+ {
+ std::lock_guard<std::mutex> lock(mutex);
+ if (status.ok() && !worker_status.ok()) {
+ status = worker_status;
+ }
+ }
+ if (remaining_workers.fetch_sub(1) == 1) {
+ // No worker or SDK callback may access the RPC controller after
on_finish runs.
+ Status final_status = check_status();
+ on_finish(final_status, sequences);
+ }
+ }
+};
+
+Status RoutineLoadTaskExecutor::_run_kinesis_scan_worker(
+ const std::shared_ptr<KinesisLatestSequenceBatch>& batch) {
+ RETURN_IF_ERROR(batch->check_status());
+ auto ctx = std::make_shared<StreamLoadContext>(_exec_env);
+ RETURN_IF_ERROR(_prepare_ctx(batch->request, ctx));
+ // Each worker owns a client. Never share or return an in-flight scan
consumer.
+ KinesisDataConsumer consumer(ctx,
config::kinesis_latest_sequence_request_timeout_ms);
+ RETURN_IF_ERROR(consumer.init(ctx));
+ while (true) {
+ RETURN_IF_ERROR(batch->check_status());
+ int index = batch->next_shard.fetch_add(1);
+ if (index >= batch->request.shard_ids_for_latest_sequences_size()) {
+ return Status::OK();
+ }
+ const auto& shard =
batch->request.shard_ids_for_latest_sequences(index);
+ std::string sequence;
+ RETURN_IF_ERROR(consumer.get_latest_sequence_number(
+ shard, [batch] { return batch->check_status(); }, &sequence));
+ std::lock_guard<std::mutex> lock(batch->mutex);
+ batch->sequences.emplace(shard, std::move(sequence));
+ }
+}
+
+void RoutineLoadTaskExecutor::get_kinesis_latest_sequence_numbers(
+ const PKinesisMetaProxyRequest& request, int64_t timeout_ms,
+ std::function<bool()> is_cancelled, KinesisScanCallback on_finish) {
+ CHECK(request.has_kinesis_info());
+ int workers = std::min(request.shard_ids_for_latest_sequences_size(),
+ config::kinesis_latest_sequence_scan_threads);
+ DCHECK_GT(workers, 0);
+ auto batch = std::make_shared<KinesisLatestSequenceBatch>(
+ request, timeout_ms,
+ [this, is_cancelled = std::move(is_cancelled)] {
+ return _kinesis_scan_stopping.load() || is_cancelled();
+ },
+ std::move(on_finish), workers);
+ for (int i = 0; i < workers; ++i) {
+ auto st = _kinesis_scan_pool->submit_func([this, batch] {
Review Comment:
[P2] Attach a non-Orphan memory limiter at this new pool-callback entry.
These workers construct a stream-load context and AWS client and can
materialize repeated 10,000-record pages, but a bare `ThreadPool` worker
initializes its TLS limiter to `Orphan`; nothing in this path switches it. That
bypasses task/subsystem accounting and trips orphan-memory checking for a
potentially large scan. Please store a per-RPC or subsystem limiter with the
batch and enter the callback with `SCOPED_ATTACH_TASK` before allocating the
context/client or fetching records.
##########
be/src/load/routine_load/routine_load_task_executor.cpp:
##########
@@ -176,13 +195,117 @@ Status
RoutineLoadTaskExecutor::get_kinesis_shard_meta(const PKinesisMetaProxyRe
std::shared_ptr<DataConsumer> consumer;
RETURN_IF_ERROR(_data_consumer_pool.get_consumer(ctx, &consumer));
- Status st =
std::static_pointer_cast<KinesisDataConsumer>(consumer)->get_shard_list(shard_ids);
+ Status st =
+
std::static_pointer_cast<KinesisDataConsumer>(consumer)->get_shard_list(shard_infos);
if (st.ok()) {
_data_consumer_pool.return_consumer(consumer);
}
return st;
}
+// All workers of one RPC share a deadline and publish results only after all
workers exit.
+struct KinesisLatestSequenceBatch {
+ PKinesisMetaProxyRequest request;
+ int64_t deadline_ms;
+ std::function<bool()> is_cancelled;
+ RoutineLoadTaskExecutor::KinesisScanCallback on_finish;
+ std::atomic<int> next_shard {0};
+ std::atomic<int> remaining_workers;
+ std::mutex mutex;
+ Status status;
+ std::map<std::string, std::string> sequences;
+
+ KinesisLatestSequenceBatch(PKinesisMetaProxyRequest req, int64_t
timeout_ms,
+ std::function<bool()> cancelled,
+ RoutineLoadTaskExecutor::KinesisScanCallback
finish, int workers)
+ : request(std::move(req)),
+ deadline_ms(timeout_ms == -1 ? -1 : MonotonicMillis() +
timeout_ms),
+ is_cancelled(std::move(cancelled)),
+ on_finish(std::move(finish)),
+ remaining_workers(workers) {}
+
+ Status check_status() {
+ std::lock_guard<std::mutex> lock(mutex);
+ RETURN_IF_ERROR(status);
+ if (is_cancelled()) {
+ return Status::Cancelled("Kinesis latest sequence scan cancelled");
+ }
+ if (deadline_ms != -1 && MonotonicMillis() >= deadline_ms) {
+ return Status::TimedOut("Kinesis latest sequence scan exceeded its
total timeout");
+ }
+ return Status::OK();
+ }
+
+ void finish_worker(const Status& worker_status) {
+ {
+ std::lock_guard<std::mutex> lock(mutex);
+ if (status.ok() && !worker_status.ok()) {
+ status = worker_status;
+ }
+ }
+ if (remaining_workers.fetch_sub(1) == 1) {
+ // No worker or SDK callback may access the RPC controller after
on_finish runs.
+ Status final_status = check_status();
+ on_finish(final_status, sequences);
+ }
+ }
+};
+
+Status RoutineLoadTaskExecutor::_run_kinesis_scan_worker(
+ const std::shared_ptr<KinesisLatestSequenceBatch>& batch) {
+ RETURN_IF_ERROR(batch->check_status());
+ auto ctx = std::make_shared<StreamLoadContext>(_exec_env);
+ RETURN_IF_ERROR(_prepare_ctx(batch->request, ctx));
+ // Each worker owns a client. Never share or return an in-flight scan
consumer.
+ KinesisDataConsumer consumer(ctx,
config::kinesis_latest_sequence_request_timeout_ms);
+ RETURN_IF_ERROR(consumer.init(ctx));
+ while (true) {
+ RETURN_IF_ERROR(batch->check_status());
+ int index = batch->next_shard.fetch_add(1);
+ if (index >= batch->request.shard_ids_for_latest_sequences_size()) {
+ return Status::OK();
+ }
+ const auto& shard =
batch->request.shard_ids_for_latest_sequences(index);
+ std::string sequence;
+ RETURN_IF_ERROR(consumer.get_latest_sequence_number(
+ shard, [batch] { return batch->check_status(); }, &sequence));
+ std::lock_guard<std::mutex> lock(batch->mutex);
+ batch->sequences.emplace(shard, std::move(sequence));
+ }
+}
+
+void RoutineLoadTaskExecutor::get_kinesis_latest_sequence_numbers(
+ const PKinesisMetaProxyRequest& request, int64_t timeout_ms,
+ std::function<bool()> is_cancelled, KinesisScanCallback on_finish) {
+ CHECK(request.has_kinesis_info());
+ int workers = std::min(request.shard_ids_for_latest_sequences_size(),
+ config::kinesis_latest_sequence_scan_threads);
+ DCHECK_GT(workers, 0);
+ auto batch = std::make_shared<KinesisLatestSequenceBatch>(
+ request, timeout_ms,
+ [this, is_cancelled = std::move(is_cancelled)] {
+ return _kinesis_scan_stopping.load() || is_cancelled();
+ },
+ std::move(on_finish), workers);
+ for (int i = 0; i < workers; ++i) {
+ auto st = _kinesis_scan_pool->submit_func([this, batch] {
+ Status worker_status;
+ try {
+ worker_status = _run_kinesis_scan_worker(batch);
+ } catch (const Exception& e) {
+ worker_status = Status::Error<false>(e.code(), e.to_string());
+ } catch (const std::exception& e) {
+ worker_status =
+ Status::InternalError("Kinesis latest sequence scan
failed: {}", e.what());
+ }
+ batch->finish_worker(worker_status);
+ });
+ if (!st.ok()) {
+ batch->finish_worker(st);
Review Comment:
[P1] Do not account this worker as rejected solely from `submit_func()`'s
return value. `ThreadPool::do_submit` queues the runnable before attempting
thread creation, and if creation fails with no workers it returns that error
without removing the queued runnable; another caller explicitly documents this
retained-runnable contract. Here the error path decrements `remaining_workers`,
so a one-worker RPC can run `done` immediately and later execute the retained
lambda, which accesses the captured controller and decrements the batch a
second time. Please add an exactly-once admission/completion guard (or use an
API with unambiguous rejection semantics) so the submitter and runnable cannot
finish the same logical worker.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java:
##########
@@ -699,82 +904,86 @@ public void modifyProperties(AlterRoutineLoadCommand
command) throws UserExcepti
private void modifyPropertiesInternal(Map<String, String> jobProperties,
KinesisDataSourceProperties
dataSourceProperties)
throws UserException {
- if (dataSourceProperties != null) {
- List<Pair<String, String>> shardPositions = Lists.newArrayList();
- Map<String, String> customKinesisProperties = Maps.newHashMap();
- boolean resetProgress = false;
- boolean hasExplicitShardPositions = false;
+ List<Pair<String, String>> shardPositions = Lists.newArrayList();
+ Map<String, String> customKinesisProperties = Maps.newHashMap();
+ boolean resetProgress = false;
+ boolean sourceChanged = false;
+ boolean sourceGenerationBumped = false;
+ boolean hasExplicitShardPositions = false;
+ if (dataSourceProperties != null) {
if
(MapUtils.isNotEmpty(dataSourceProperties.getOriginalDataSourceProperties())) {
shardPositions =
dataSourceProperties.getKinesisShardPositions();
customKinesisProperties =
dataSourceProperties.getCustomKinesisProperties();
hasExplicitShardPositions = !shardPositions.isEmpty();
+ sourceChanged = true;
}
+ resetProgress =
!Strings.isNullOrEmpty(dataSourceProperties.getStream());
+ sourceChanged |= resetProgress
+ || !Strings.isNullOrEmpty(dataSourceProperties.getRegion())
+ ||
!Strings.isNullOrEmpty(dataSourceProperties.getEndpoint());
+ }
- // Update custom properties
+ // Validate every failure-prone input before mutating Kinesis or
common job state.
+ if (hasExplicitShardPositions && !resetProgress) {
+ ((KinesisProgress) progress).checkShards(shardPositions);
+ }
+ if (!jobProperties.isEmpty()) {
+ Map<String, String> copiedJobProperties =
Maps.newHashMap(jobProperties);
+ modifyCommonJobProperties(copiedJobProperties);
+ this.jobProperties.putAll(copiedJobProperties);
+ if
(jobProperties.containsKey(CreateRoutineLoadInfo.PARTIAL_COLUMNS)) {
+ this.isPartialUpdate =
BooleanUtils.toBoolean(jobProperties.get(CreateRoutineLoadInfo.PARTIAL_COLUMNS));
+ }
+ if
(jobProperties.containsKey(CreateRoutineLoadInfo.PARTIAL_UPDATE_NEW_KEY_POLICY))
{
+ String policy =
jobProperties.get(CreateRoutineLoadInfo.PARTIAL_UPDATE_NEW_KEY_POLICY);
+ this.partialUpdateNewKeyPolicy =
"ERROR".equalsIgnoreCase(policy)
+ ? TPartialUpdateNewRowPolicy.ERROR :
TPartialUpdateNewRowPolicy.APPEND;
+ }
+ }
+
+ if (dataSourceProperties != null) {
if (!customKinesisProperties.isEmpty()) {
this.customProperties.putAll(customKinesisProperties);
convertCustomProperties(true);
}
-
- // Modify stream if provided
if (!Strings.isNullOrEmpty(dataSourceProperties.getStream())) {
this.stream = dataSourceProperties.getStream();
- resetProgress = true;
}
-
- // Modify region if provided
if (!Strings.isNullOrEmpty(dataSourceProperties.getRegion())) {
this.region = dataSourceProperties.getRegion();
}
-
- // Modify endpoint if provided
if (!Strings.isNullOrEmpty(dataSourceProperties.getEndpoint())) {
this.endpoint = dataSourceProperties.getEndpoint();
}
+ if (sourceChanged && !sourceGenerationBumped) {
+ sourceGeneration++;
+ resetLatestSequenceFetch();
+ newCurrentKinesisShardInfos = null;
+ sourceGenerationBumped = true;
+ }
if (resetProgress) {
this.progress = new KinesisProgress();
+ this.shardTopology.reset();
Review Comment:
[P1] Use the full source-identity change here, not only `resetProgress`
(which is true only when `stream` is supplied). The new code correctly treats
region and endpoint changes as `sourceChanged`, but a region-only or
endpoint-only ALTER retains the previous topology, completion states,
explicit-shard filter, lag, and sequence numbers. Kinesis shard IDs are reused
across sources, so the next task can send an old source's sequence as
`AFTER_SEQUENCE_NUMBER` to the new source or leave a same-named shard
permanently completed. Please reset all source-bound state for
stream/region/endpoint changes and then apply any explicit new positions; cover
region-only and endpoint-only ALTER plus journal replay.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
