Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

c/topic_table: notify ntp delta waiters in batches #24761

Open
wants to merge 1 commit into
base: dev
Choose a base branch
from
Open

c/topic_table: notify ntp delta waiters in batches #24761

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
161 changes: 79 additions & 82 deletions src/v/cluster/topic_table.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,8 +44,7 @@ topic_table::apply(create_topic_cmd cmd, model::offset offset) {
_last_applied_revision_id = model::revision_id(offset);
if (_topics.contains(cmd.key)) {
// topic already exists
return ss::make_ready_future<std::error_code>(
errc::topic_already_exists);
co_return errc::topic_already_exists;
}

const auto migration_state = _migrated_resources.get_topic_state(cmd.key);
Expand All @@ -54,18 +53,15 @@ topic_table::apply(create_topic_cmd cmd, model::offset offset) {
&& migration_state
!= data_migrations::migrated_resource_state::non_restricted) {
vlog(clusterlog.debug, "topic {} already migrated", cmd.key);
return ss::make_ready_future<std::error_code>(
errc::topic_already_exists);
co_return errc::topic_already_exists;
}

if (!schema_id_validation_validator::is_valid(cmd.value.cfg.properties)) {
return ss::make_ready_future<std::error_code>(
schema_id_validation_validator::ec);
co_return schema_id_validation_validator::ec;
}

if (!topic_multi_property_validation(cmd.value.cfg.properties)) {
return ss::make_ready_future<std::error_code>(
errc::topic_invalid_config);
co_return errc::topic_invalid_config;
}

std::optional<model::initial_revision_id> remote_revision
Expand Down Expand Up @@ -111,10 +107,11 @@ topic_table::apply(create_topic_cmd cmd, model::offset offset) {
std::move(md),
});
_topics_map_revision++;
notify_waiters();

_probe.handle_topic_creation(std::move(cmd.key));
return ss::make_ready_future<std::error_code>(errc::success);

co_await notify_waiters();

co_return errc::success;
}

ss::future<> topic_table::stop() { return ss::now(); }
Expand All @@ -123,47 +120,49 @@ ss::future<std::error_code>
topic_table::apply(delete_topic_cmd cmd, model::offset offset) {
_last_applied_revision_id = model::revision_id(offset);

co_return do_local_delete(cmd.key, offset, false);
co_return co_await do_local_delete(cmd.key, offset, false);
}

std::error_code topic_table::do_local_delete(
ss::future<std::error_code> topic_table::do_local_delete(
model::topic_namespace nt, model::offset offset, bool ignore_migration) {
const auto migration_state = _migrated_resources.get_topic_state(nt);
if (
!ignore_migration
&& migration_state
!= data_migrations::migrated_resource_state::non_restricted) {
return errc::resource_is_being_migrated;
co_return errc::resource_is_being_migrated;
}
auto tp = _topics.find(nt);
if (tp == _topics.end()) {
co_return errc::topic_not_exists;
}
if (auto tp = _topics.find(nt); tp != _topics.end()) {
_pending_topic_deltas.emplace_back(
tp->second.get_revision(),
nt,
model::revision_id{offset},
topic_table_topic_delta_type::removed);

for (auto& [_, p_as] : tp->second.get_assignments()) {
_partition_count--;
auto ntp = model::ntp(nt.ns, nt.tp, p_as.id);
_updates_in_progress.erase(ntp);
on_partition_deletion(ntp);
_pending_ntp_deltas.emplace_back(
std::move(ntp),
p_as.group,
model::revision_id(offset),
topic_table_ntp_delta_type::removed);
}

_topics.erase(tp);
_disabled_partitions.erase(nt);
_topics_map_revision++;
notify_waiters();
_probe.handle_topic_deletion(nt);
_pending_topic_deltas.emplace_back(
tp->second.get_revision(),
nt,
model::revision_id{offset},
topic_table_topic_delta_type::removed);

return errc::success;
for (auto& [_, p_as] : tp->second.get_assignments()) {
_partition_count--;
auto ntp = model::ntp(nt.ns, nt.tp, p_as.id);
_updates_in_progress.erase(ntp);
on_partition_deletion(ntp);
_pending_ntp_deltas.emplace_back(
std::move(ntp),
p_as.group,
model::revision_id(offset),
topic_table_ntp_delta_type::removed);
}

return errc::topic_not_exists;
_topics.erase(tp);
_disabled_partitions.erase(nt);
_topics_map_revision++;
_probe.handle_topic_deletion(nt);

co_await notify_waiters();

co_return errc::success;
}

ss::future<std::error_code>
Expand Down Expand Up @@ -223,10 +222,10 @@ topic_table::apply(topic_lifecycle_transition soft_del, model::offset offset) {
}
case topic_lifecycle_transition_mode::oneshot_delete:
case topic_lifecycle_transition_mode::delete_migrated:
return ssx::now(do_local_delete(
return do_local_delete(
soft_del.topic.nt,
offset,
soft_del.mode == topic_lifecycle_transition_mode::delete_migrated));
soft_del.mode == topic_lifecycle_transition_mode::delete_migrated);
case topic_lifecycle_transition_mode::purged:
switch (soft_del.domain) {
case topic_purge_domain::cloud_storage: {
Expand Down Expand Up @@ -344,7 +343,7 @@ topic_table::apply(create_partition_cmd cmd, model::offset offset) {
model::revision_id(offset),
topic_table_ntp_delta_type::added);
}
notify_waiters();
co_await notify_waiters();
co_return errc::success;
}

Expand All @@ -369,23 +368,22 @@ ss::future<std::error_code> topic_table::do_apply(

auto tp = _topics.find(model::topic_namespace_view(cmd_data.ntp));
if (tp == _topics.end()) {
return ss::make_ready_future<std::error_code>(errc::topic_not_exists);
co_return errc::topic_not_exists;
}

auto current_assignment_it = tp->second.get_assignments().find(
cmd_data.ntp.tp.partition);

if (current_assignment_it == tp->second.get_assignments().end()) {
return ss::make_ready_future<std::error_code>(
errc::partition_not_exists);
co_return errc::partition_not_exists;
}

if (is_disabled(cmd_data.ntp)) {
return ss::make_ready_future<std::error_code>(errc::partition_disabled);
co_return errc::partition_disabled;
}

if (_updates_in_progress.contains(cmd_data.ntp)) {
return ss::make_ready_future<std::error_code>(errc::update_in_progress);
co_return errc::update_in_progress;
}

change_partition_replicas(
Expand All @@ -395,9 +393,9 @@ ss::future<std::error_code> topic_table::do_apply(
o,
false,
cmd_data.policy);
notify_waiters();
co_await notify_waiters();

return ss::make_ready_future<std::error_code>(errc::success);
co_return errc::success;
}

static replicas_revision_map update_replicas_revisions(
Expand Down Expand Up @@ -429,32 +427,28 @@ topic_table::apply(finish_moving_partition_replicas_cmd cmd, model::offset o) {
_last_applied_revision_id = model::revision_id(o);
auto tp = _topics.find(model::topic_namespace_view(cmd.key));
if (tp == _topics.end()) {
return ss::make_ready_future<std::error_code>(errc::topic_not_exists);
co_return errc::topic_not_exists;
}

// calculate deleta for backend
auto current_assignment_it = tp->second.get_assignments().find(
cmd.key.tp.partition);

if (current_assignment_it == tp->second.get_assignments().end()) {
return ss::make_ready_future<std::error_code>(
errc::partition_not_exists);
co_return errc::partition_not_exists;
}

if (current_assignment_it->second.replicas != cmd.value) {
return ss::make_ready_future<std::error_code>(
errc::invalid_node_operation);
co_return errc::invalid_node_operation;
}
auto it = _updates_in_progress.find(cmd.key);
if (it == _updates_in_progress.end()) {
return ss::make_ready_future<std::error_code>(
errc::no_update_in_progress);
co_return errc::no_update_in_progress;
}

auto p_meta_it = tp->second.partitions.find(cmd.key.tp.partition);
if (p_meta_it == tp->second.partitions.end()) {
return ss::make_ready_future<std::error_code>(
errc::partition_not_exists);
co_return errc::partition_not_exists;
}
if (!is_cancelled_state(it->second.get_state())) {
// update went through and the cancellation didn't happen, we must
Expand All @@ -479,9 +473,9 @@ topic_table::apply(finish_moving_partition_replicas_cmd cmd, model::offset o) {
model::revision_id(o),
topic_table_ntp_delta_type::replicas_updated);

notify_waiters();
co_await notify_waiters();

return ss::make_ready_future<std::error_code>(errc::success);
co_return errc::success;
}

ss::future<std::error_code>
Expand Down Expand Up @@ -551,7 +545,7 @@ topic_table::apply(cancel_moving_partition_replicas_cmd cmd, model::offset o) {
current_assignment_it->second.group,
model::revision_id(o),
topic_table_ntp_delta_type::replicas_updated);
notify_waiters();
co_await notify_waiters();

co_return errc::success;
}
Expand Down Expand Up @@ -624,7 +618,7 @@ topic_table::apply(revert_cancel_partition_move_cmd cmd, model::offset o) {
current_assignment_it->second.group,
model::revision_id(o),
topic_table_ntp_delta_type::replicas_updated);
notify_waiters();
co_await notify_waiters();

co_return errc::success;
}
Expand Down Expand Up @@ -692,7 +686,7 @@ topic_table::apply(move_topic_replicas_cmd cmd, model::offset o) {
reconfiguration_policy::full_local_retention);
}

notify_waiters();
co_await notify_waiters();

co_return errc::success;
}
Expand All @@ -703,24 +697,23 @@ topic_table::apply(force_partition_reconfiguration_cmd cmd, model::offset o) {
// Check the topic exists.
auto tp = _topics.find(model::topic_namespace_view(cmd.key));
if (tp == _topics.end()) {
return ss::make_ready_future<std::error_code>(errc::topic_not_exists);
co_return errc::topic_not_exists;
}

auto current_assignment_it = tp->second.get_assignments().find(
cmd.key.tp.partition);

if (current_assignment_it == tp->second.get_assignments().end()) {
return ss::make_ready_future<std::error_code>(
errc::partition_not_exists);
co_return errc::partition_not_exists;
}

if (is_disabled(cmd.key)) {
return ss::make_ready_future<std::error_code>(errc::partition_disabled);
co_return errc::partition_disabled;
}

if (auto it = _updates_in_progress.find(cmd.key);
it != _updates_in_progress.end()) {
return ss::make_ready_future<std::error_code>(errc::update_in_progress);
co_return errc::update_in_progress;
}

change_partition_replicas(
Expand All @@ -734,9 +727,10 @@ topic_table::apply(force_partition_reconfiguration_cmd cmd, model::offset o) {
* reconfiguring partition.
*/
reconfiguration_policy::full_local_retention);
notify_waiters();

return ss::make_ready_future<std::error_code>(errc::success);
co_await notify_waiters();

co_return errc::success;
}

ss::future<std::error_code>
Expand Down Expand Up @@ -804,7 +798,7 @@ topic_table::apply(set_topic_partitions_disabled_cmd cmd, model::offset o) {
}

_topics_map_revision++;
notify_waiters();
co_await notify_waiters();

co_return errc::success;
}
Expand Down Expand Up @@ -1162,7 +1156,7 @@ topic_table::apply(update_topic_properties_cmd cmd, model::offset o) {
topic_table_ntp_delta_type::properties_updated);
}

notify_waiters();
co_await notify_waiters();

co_return make_error_code(errc::success);
}
Expand Down Expand Up @@ -1617,7 +1611,7 @@ ss::future<> topic_table::apply_snapshot(
}

// 3. notify delta waiters
notify_waiters();
co_await notify_waiters();

_last_applied_revision_id = snap_revision;
}
Expand All @@ -1636,24 +1630,27 @@ void topic_table::reset_partitions_to_force_reconfigure(
_partitions_to_force_reconfigure_revision++;
}

void topic_table::notify_waiters() {
// \ref notify_waiters is called after every apply. Hence for the most
// part there should only be a few items in \ref _pending_deltas that need
// to be sent to callbacks in \ref notifications.

ss::future<> topic_table::notify_waiters() {
if (!_pending_topic_deltas.empty()) {
for (auto& cb : _topic_notifications) {
cb.second(_pending_topic_deltas);
}
}
_pending_topic_deltas.clear();

ntp_delta_range_t changes{
_pending_ntp_deltas.cbegin(), _pending_ntp_deltas.cend()};
if (!changes.empty()) {
const ssize_t batch_size = 128;
for (size_t i = 0; i < _pending_ntp_deltas.size(); i += batch_size) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit. Have you checked whether it's okay to increment this iterator past the end? From what I can see, it's down to the iterator implementation, so it's not guaranteed by the iterator traits. Currently iterator position is stored as an index, so it looks safe. But if we change it to e.g. a pair of chunk_id and pos_in_chunk, this code may become invalid. WDYT?

auto begin = _pending_ntp_deltas.begin() + i;
auto end = _pending_ntp_deltas.end();
if (end - begin > batch_size) {
end = begin + batch_size;
}
Comment on lines +1644 to +1647
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Have you tried auto end = std::min(begin + batch_size, _pending_ntp_deltas.end())? According to the iterator category tag It should just work.


for (auto& cb : _ntp_notifications) {
cb.second(changes);
cb.second(ntp_delta_range_t{begin, end});
}

co_await ss::coroutine::maybe_yield();
}
_pending_ntp_deltas.clear();

Expand Down
4 changes: 2 additions & 2 deletions src/v/cluster/topic_table.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -672,7 +672,7 @@ class topic_table {
uint64_t id;
};

void notify_waiters();
ss::future<> notify_waiters();

void change_partition_replicas(
model::ntp ntp,
Expand All @@ -684,7 +684,7 @@ class topic_table {

class snapshot_applier;

std::error_code do_local_delete(
ss::future<std::error_code> do_local_delete(
model::topic_namespace nt, model::offset offset, bool ignore_migration);
ss::future<std::error_code>
do_apply(update_partition_replicas_cmd_data, model::offset);
Expand Down
Loading