wgtmac commented on code in PR #1033:
URL: https://github.com/apache/iceberg-cpp/pull/1033#discussion_r4305273988
##########
src/iceberg/catalog/rest/auth/auth_manager.cc:
##########
@@ -319,12 +323,14 @@ class OAuth2Manager : public AuthManager {
std::optional<OAuthTokenResponse> auth_response_;
std::optional<std::chrono::steady_clock::time_point> start_time_;
std::shared_ptr<HttpClient> refresh_client_;
+ std::unique_ptr<::iceberg::internal::ScheduledExecutor> scheduler_;
Review Comment:
Fixed. Each OAuth2 session now retains the scheduled-executor state used by
its refresh tasks, so resetting or destroying the manager does not invalidate a
session's scheduler. The injected Executor remains borrowed and its lifetime is
documented.
##########
src/iceberg/catalog/rest/auth/auth_session_internal.h:
##########
@@ -118,10 +119,10 @@ class OAuth2Session final : public AuthSession,
if (!closed_.compare_exchange_strong(expected, true)) {
return {};
}
- TokenRefreshScheduler::Instance().Cancel(scheduled_task_id_.exchange(0));
+ scheduled_task_.Cancel();
Review Comment:
Fixed. Scheduled-task handle access is now protected by a dedicated mutex.
Close marks the session closed before cancelling, and any task assigned
concurrently is cancelled by the guarded store path.
##########
src/iceberg/util/scheduled_executor_internal.h:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <chrono>
+#include <condition_variable>
+#include <cstdint>
+#include <mutex>
+#include <queue>
+#include <thread>
+#include <unordered_set>
+#include <vector>
+
+#include "iceberg/result.h"
+#include "iceberg/util/executor.h"
+
+namespace iceberg::internal {
+
+class ScheduledExecutor;
+
+class ScheduledTask {
+ public:
+ ScheduledTask() = default;
+ ScheduledTask(ScheduledExecutor* executor, uint64_t id)
+ : executor_(executor), id_(id) {}
+
+ void Cancel();
+ bool valid() const { return executor_ != nullptr && id_ != 0; }
+
+ private:
+ ScheduledExecutor* executor_ = nullptr;
+ uint64_t id_ = 0;
+};
+
+/// A delay queue that dispatches due tasks to an existing Executor.
+///
+/// The executor is borrowed and must outlive this object and all scheduled
tasks.
+class ScheduledExecutor {
Review Comment:
Fixed. ScheduledExecutor and ScheduledTask are now exported so their
out-of-line symbols remain available when the core and REST libraries are built
as shared libraries.
##########
src/iceberg/catalog/rest/auth/auth_manager.h:
##########
@@ -38,6 +38,10 @@ class ICEBERG_REST_EXPORT AuthManager {
public:
virtual ~AuthManager() = default;
+ /// Configure the borrowed executor used for background authentication
refresh.
+ /// The executor must outlive this manager and all sessions it creates.
+ virtual void SetAuthRefreshExecutor(Executor* /*executor*/) {}
Review Comment:
Fixed. SigV4AuthManager now forwards SetAuthRefreshExecutor to its delegate,
so OAuth2 refresh remains enabled for SigV4-over-OAuth2 catalogs.
##########
src/iceberg/util/scheduled_executor.cc:
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.
+ */
+
+#include <optional>
+#include <utility>
+
+#include "iceberg/util/scheduled_executor_internal.h"
+
+namespace iceberg::internal {
+
+void ScheduledTask::Cancel() {
+ if (valid()) {
+ executor_->Cancel(id_);
+ executor_ = nullptr;
+ id_ = 0;
+ }
+}
+
+ScheduledExecutor::ScheduledExecutor(Executor& executor)
+ : executor_(executor), timer_thread_([this] { Run(); }) {}
+
+ScheduledExecutor::~ScheduledExecutor() { Shutdown(); }
+
+ScheduledTask
ScheduledExecutor::Schedule(std::chrono::steady_clock::time_point deadline,
+ ExecutorTask task) {
+ std::lock_guard lock(mutex_);
+ if (shutdown_) return {};
+ auto id = next_id_++;
+ tasks_.push(Task{.id = id, .deadline = deadline, .task = std::move(task)});
+ scheduled_.insert(id);
+ cv_.notify_one();
+ return ScheduledTask(this, id);
+}
+
+ScheduledTask ScheduledExecutor::Schedule(std::chrono::milliseconds delay,
+ ExecutorTask task) {
+ return Schedule(std::chrono::steady_clock::now() + delay, std::move(task));
+}
+
+void ScheduledExecutor::Cancel(uint64_t id) {
+ if (id == 0) return;
+ {
+ std::lock_guard lock(mutex_);
+ if (scheduled_.erase(id) != 0) cancelled_.insert(id);
+ }
+ cv_.notify_one();
+}
+
+void ScheduledExecutor::Shutdown() {
+ {
+ std::lock_guard lock(mutex_);
+ if (shutdown_) return;
+ shutdown_ = true;
+ while (!tasks_.empty()) tasks_.pop();
+ cancelled_.clear();
+ scheduled_.clear();
+ }
+ cv_.notify_one();
+ if (timer_thread_.joinable()) timer_thread_.join();
+}
+
+void ScheduledExecutor::Run() {
+ while (true) {
+ std::optional<Task> task;
+ {
+ std::unique_lock lock(mutex_);
+ cv_.wait(lock, [this] { return shutdown_ || !tasks_.empty(); });
+ if (shutdown_) return;
+
+ while (!tasks_.empty() && cancelled_.contains(tasks_.top().id)) {
+ cancelled_.erase(tasks_.top().id);
+ tasks_.pop();
+ }
+ if (tasks_.empty()) continue;
+
+ auto deadline = tasks_.top().deadline;
+ if (cv_.wait_until(lock, deadline, [this, deadline] {
+ return shutdown_ || tasks_.empty() || tasks_.top().deadline !=
deadline ||
+ cancelled_.contains(tasks_.top().id);
+ })) {
+ continue;
+ }
+
+ task.emplace(std::move(const_cast<Task&>(tasks_.top())));
+ tasks_.pop();
+ scheduled_.erase(task->id);
+ if (cancelled_.erase(task->id) != 0) continue;
+ }
+ std::ignore = executor_.Submit(std::move(task->task));
Review Comment:
Fixed. ScheduledExecutor now accepts a rejection callback for
Executor::Submit failures. OAuth2 refresh tasks use it to reschedule through
the session instead of silently stopping after dispatch rejection.
--
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]
