diff --git a/src/v/cluster/scheduling/tests/BUILD b/src/v/cluster/scheduling/tests/BUILD index 75bed9f9408a1..8992c394c8a94 100644 --- a/src/v/cluster/scheduling/tests/BUILD +++ b/src/v/cluster/scheduling/tests/BUILD @@ -1,5 +1,24 @@ load("//bazel:test.bzl", "redpanda_cc_gtest") +redpanda_cc_gtest( + name = "allocated_partition_test", + timeout = "short", + srcs = [ + "allocated_partition_test.cc", + ], + deps = [ + "//src/v/cluster:errc", + "//src/v/cluster:scheduling_allocation", + "//src/v/config", + "//src/v/container:chunked_vector", + "//src/v/features", + "//src/v/model", + "//src/v/test_utils:gtest", + "@googletest//:gtest", + "@seastar", + ], +) + redpanda_cc_gtest( name = "leader_pinning_test", timeout = "short", diff --git a/src/v/cluster/scheduling/tests/allocated_partition_test.cc b/src/v/cluster/scheduling/tests/allocated_partition_test.cc new file mode 100644 index 0000000000000..757613ca67459 --- /dev/null +++ b/src/v/cluster/scheduling/tests/allocated_partition_test.cc @@ -0,0 +1,335 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Use of this software is governed by the Business Source License + * included in the file licenses/BSL.md + * + * As of the Change Date specified in that file, in accordance with + * the Business Source License, use of this software will be governed + * by the Apache License, Version 2.0 + */ + +#include "cluster/errc.h" +#include "cluster/scheduling/allocation_node.h" +#include "cluster/scheduling/allocation_state.h" +#include "cluster/scheduling/types.h" +#include "config/mock_property.h" +#include "config/property.h" +#include "container/chunked_vector.h" +#include "features/feature_table.h" +#include "model/fundamental.h" +#include "model/metadata.h" +#include "model/namespace.h" + +#include + +#include + +#include +#include +#include +#include +#include +#include + +namespace cluster { + +class allocated_partition_test : public ::testing::Test { +protected: + void fresh_partition_has_no_changes(); + void noop_reallocate_owns_nothing(); + void new_partition_all_replicas_are_new(); + void move_allocates_new_preserves_original(); + void try_revert_restores_original(); + void try_revert_on_fresh_partition_returns_no_update(); + void try_revert_rejects_invalid_steps(); + void destructor_reconciles_allocation_state(); + + // Move the replica on `from` to a freshly allocated one on `to`; returns + // the step so it can be reverted. + reallocation_step move( + allocated_partition& partition, model::node_id from, model::node_id to) { + auto previous = partition.prepare_move(from); + std::optional previous_shard; + if (previous) { + previous_shard = previous->bs; + } + return reallocation_step( + partition.add_replica(to, previous), previous_shard); + } + + static model::broker_shard + make_broker_shard(model::node_id node, uint32_t shard = 0) { + return model::broker_shard{node, shard}; + } + + // node ids [1, count]. + static std::vector node_id_range(int count) { + std::vector values(count); + std::ranges::iota(values, 1); + std::vector ids; + ids.reserve(values.size()); + std::ranges::transform(values, std::back_inserter(ids), [](int value) { + return model::node_id(value); + }); + return ids; + } + + static std::vector + broker_shards(const std::vector& ids) { + std::vector shards; + shards.reserve(ids.size()); + std::ranges::transform( + ids, std::back_inserter(shards), [](model::node_id id) { + return make_broker_shard(id); + }); + return shards; + } + + void register_node_range(const std::vector& ids) { + for (auto id : ids) { + _state.register_node( + std::make_unique( + id, + /*cpus=*/4, + config::mock_binding(_partitions_per_shard), + _partitions_reserve_shard0.bind(), + _internal_topics.bind())); + } + } + + // Reflect an already-existing partition's replicas in the allocation-state + // counters, mirroring replicas a previous allocation request placed. + void add_existing(const std::vector& replicas) { + for (const auto& replica : replicas) { + _state.add_allocation(replica); + _state.add_final_count(replica); + } + } + + const allocation_node& node(model::node_id id) const { + return *_state.allocation_nodes().at(id); + } + uint32_t allocated(model::node_id id) const { + return node(id).allocated_partitions()(); + } + uint32_t final_count(model::node_id id) const { + return node(id).final_partitions()(); + } + + static bool + has_replica(const allocated_partition& partition, model::node_id id) { + return std::ranges::any_of( + partition.replicas(), + [id](const auto& replica) { return replica.node_id == id; }); + } + + model::ntp test_ntp{ + model::kafka_namespace, model::topic("test"), model::partition_id(0)}; + config::mock_property> _internal_topics{{}}; + config::mock_property _partitions_reserve_shard0{0}; + uint32_t _partitions_per_shard{1000}; + features::feature_table _features; + allocation_state _state{ + _features, + config::mock_binding(_partitions_per_shard), + _partitions_reserve_shard0.bind(), + _internal_topics.bind()}; +}; + +void allocated_partition_test::fresh_partition_has_no_changes() { + auto nodes = node_id_range(3); + register_node_range(nodes); + + allocated_partition partition(test_ntp, broker_shards(nodes), _state); + + EXPECT_FALSE(partition.has_changes()); + EXPECT_EQ(partition.replicas().size(), 3u); + // Without a snapshot, is_original() falls back to scanning current + // replicas. + EXPECT_TRUE(partition.is_original(model::node_id(1))); + EXPECT_FALSE(partition.is_original(model::node_id(9))); +} + +// The no-op reallocate: add_replica() never runs, so no snapshot is taken and +// the pre-existing replicas must NOT be reported as newly allocated. Regression +// test for the allocation-counter underflow fix. +void allocated_partition_test::noop_reallocate_owns_nothing() { + auto nodes = node_id_range(3); + register_node_range(nodes); + auto replicas = broker_shards(nodes); + add_existing(replicas); + + allocated_partition partition(test_ntp, replicas, _state); + chunked_vector added; + auto released = partition.release_new_partition(added); + + EXPECT_TRUE(added.empty()); + EXPECT_EQ(released.size(), 3u); + for (auto id : nodes) { + EXPECT_EQ(allocated(id), 1u); + EXPECT_EQ(final_count(id), 1u); + } +} + +// A brand-new partition allocated from zero existing replicas: the snapshot is +// present but empty, so every replica is newly allocated and owned by these +// units. +void allocated_partition_test::new_partition_all_replicas_are_new() { + auto nodes = node_id_range(3); + register_node_range(nodes); + + allocated_partition partition(test_ntp, {}, _state); + std::ranges::for_each(nodes, [&](model::node_id id) { + partition.add_replica(id, std::nullopt); + }); + + EXPECT_TRUE(partition.has_changes()); + EXPECT_FALSE(partition.is_original(model::node_id(1))); + for (auto id : nodes) { + EXPECT_EQ(allocated(id), 1u); + EXPECT_EQ(final_count(id), 1u); + } + + chunked_vector added; + auto released = partition.release_new_partition(added); + EXPECT_EQ(added.size(), 3u); + EXPECT_EQ(released.size(), 3u); +} + +// Moving a replica preserves the original replica's allocation (its shard is +// kept and its allocation is not released) and allocates a new one on the +// target node. +void allocated_partition_test::move_allocates_new_preserves_original() { + register_node_range(node_id_range(4)); + auto replicas = broker_shards(node_id_range(3)); + add_existing(replicas); + + allocated_partition partition(test_ntp, replicas, _state); + move(partition, model::node_id(3), model::node_id(4)); + + EXPECT_TRUE(partition.has_changes()); + EXPECT_TRUE(partition.is_original(model::node_id(3))); + EXPECT_FALSE(partition.is_original(model::node_id(4))); + EXPECT_FALSE(has_replica(partition, model::node_id(3))); + EXPECT_TRUE(has_replica(partition, model::node_id(4))); + + // node 3 keeps its allocation (still original) but loses its final count; + // node 4 gains both. + EXPECT_EQ(allocated(model::node_id(3)), 1u); + EXPECT_EQ(final_count(model::node_id(3)), 0u); + EXPECT_EQ(allocated(model::node_id(4)), 1u); + EXPECT_EQ(final_count(model::node_id(4)), 1u); +} + +// Reverting the move restores the original replica set and allocation counters +// without underflow. The snapshot latches: even though the replica set matches +// the original again, a further revert is treated as an in-progress +// reallocation. +void allocated_partition_test::try_revert_restores_original() { + register_node_range(node_id_range(4)); + auto replicas = broker_shards(node_id_range(3)); + add_existing(replicas); + + allocated_partition partition(test_ntp, replicas, _state); + auto move_step = move(partition, model::node_id(3), model::node_id(4)); + + ASSERT_EQ(partition.try_revert(move_step), errc::success); + + EXPECT_FALSE(partition.has_changes()); + EXPECT_TRUE(has_replica(partition, model::node_id(3))); + EXPECT_FALSE(has_replica(partition, model::node_id(4))); + EXPECT_EQ(allocated(model::node_id(3)), 1u); + EXPECT_EQ(final_count(model::node_id(3)), 1u); + EXPECT_EQ(allocated(model::node_id(4)), 0u); + EXPECT_EQ(final_count(model::node_id(4)), 0u); + + // Snapshot latches: node 4 is no longer a replica, so a second revert + // reports node_does_not_exists, not no_update_in_progress. + EXPECT_EQ(partition.try_revert(move_step), errc::node_does_not_exists); +} + +void allocated_partition_test:: + try_revert_on_fresh_partition_returns_no_update() { + register_node_range(node_id_range(4)); + allocated_partition partition( + test_ntp, broker_shards(node_id_range(3)), _state); + + reallocation_step fabricated_step( + make_broker_shard(model::node_id(4)), + make_broker_shard(model::node_id(3))); + EXPECT_EQ( + partition.try_revert(fabricated_step), errc::no_update_in_progress); +} + +void allocated_partition_test::try_revert_rejects_invalid_steps() { + register_node_range(node_id_range(4)); + auto replicas = broker_shards(node_id_range(3)); + add_existing(replicas); + + allocated_partition partition(test_ntp, replicas, _state); + auto move_step = move(partition, model::node_id(3), model::node_id(4)); + + // current() node isn't a replica. + EXPECT_EQ( + partition.try_revert( + reallocation_step(make_broker_shard(model::node_id(9)), std::nullopt)), + errc::node_does_not_exists); + + // previous() node is already a replica. + EXPECT_EQ( + partition.try_revert(reallocation_step( + move_step.current(), make_broker_shard(model::node_id(1)))), + errc::invalid_request); +} + +// Dropping an allocated_partition without releasing it must free the replicas +// it allocated and restore the final counts of originals that were moved away, +// leaving the allocation state exactly as it was before. +void allocated_partition_test::destructor_reconciles_allocation_state() { + register_node_range(node_id_range(4)); + auto existing = node_id_range(3); + add_existing(broker_shards(existing)); + + { + allocated_partition partition( + test_ntp, broker_shards(existing), _state); + move(partition, model::node_id(3), model::node_id(4)); + EXPECT_EQ(allocated(model::node_id(4)), 1u); + EXPECT_EQ(final_count(model::node_id(3)), 0u); + } + + for (auto id : existing) { + EXPECT_EQ(allocated(id), 1u); + EXPECT_EQ(final_count(id), 1u); + } + EXPECT_EQ(allocated(model::node_id(4)), 0u); + EXPECT_EQ(final_count(model::node_id(4)), 0u); +} + +TEST_F(allocated_partition_test, FreshPartitionHasNoChanges) { + fresh_partition_has_no_changes(); +} +TEST_F(allocated_partition_test, NoopReallocateOwnsNothing) { + noop_reallocate_owns_nothing(); +} +TEST_F(allocated_partition_test, NewPartitionAllReplicasAreNew) { + new_partition_all_replicas_are_new(); +} +TEST_F(allocated_partition_test, MoveAllocatesNewPreservesOriginal) { + move_allocates_new_preserves_original(); +} +TEST_F(allocated_partition_test, TryRevertRestoresOriginal) { + try_revert_restores_original(); +} +TEST_F(allocated_partition_test, TryRevertOnFreshPartitionReturnsNoUpdate) { + try_revert_on_fresh_partition_returns_no_update(); +} +TEST_F(allocated_partition_test, TryRevertRejectsInvalidSteps) { + try_revert_rejects_invalid_steps(); +} +TEST_F(allocated_partition_test, DestructorReconcilesAllocationState) { + destructor_reconciles_allocation_state(); +} + +} // namespace cluster diff --git a/src/v/cluster/scheduling/types.cc b/src/v/cluster/scheduling/types.cc index 74e1fe0d0844a..48ef9c9b64431 100644 --- a/src/v/cluster/scheduling/types.cc +++ b/src/v/cluster/scheduling/types.cc @@ -12,6 +12,7 @@ #include "cluster/scheduling/types.h" #include "base/format_to.h" +#include "base/vassert.h" #include "cluster/scheduling/allocation_state.h" #include "utils/exceptions.h" @@ -59,6 +60,33 @@ allocated_partition::allocated_partition( , _replicas(std::move(replicas)) , _state(state.weak_from_this()) {} +bool allocated_partition::original_node2shard::has_ever_seen_modifications() + const { + return _snapshot.has_value(); +} + +void allocated_partition::original_node2shard::capture( + const replicas_t& replicas) { + vassert(!_snapshot, "original placement already captured"); + _snapshot.emplace(); + for (const auto& bs : replicas) { + _snapshot->emplace(bs.node_id, bs.shard); + } +} + +void allocated_partition::original_node2shard::reset() { _snapshot.reset(); } + +bool allocated_partition::original_node2shard::is_original( + model::node_id node) const { + vassert(_snapshot, "original placement snapshot not captured"); + return _snapshot->contains(node); +} + +size_t allocated_partition::original_node2shard::size() const { + vassert(_snapshot, "original placement snapshot not captured"); + return _snapshot->size(); +} + std::optional allocated_partition::prepare_move(model::node_id prev_node) const { previous_replica prev; @@ -81,15 +109,12 @@ model::broker_shard allocated_partition::add_replica( "allocation_state was concurrently replaced"); } - if (!_original_node2shard) { - _original_node2shard.emplace(); - for (const auto& bs : _replicas) { - _original_node2shard->emplace(bs.node_id, bs.shard); - } + if (!_original_node2shard.has_ever_seen_modifications()) { + _original_node2shard.capture(_replicas); } if (prev) { - if (!_original_node2shard->contains(prev->bs.node_id)) { + if (!_original_node2shard.is_original(prev->bs.node_id)) { _state->remove_allocation(prev->bs); } _state->remove_final_count(prev->bs); @@ -97,8 +122,8 @@ model::broker_shard allocated_partition::add_replica( model::broker_shard replica{.node_id = node}; if ( - auto it = _original_node2shard->find(node); - it != _original_node2shard->end()) { + auto it = _original_node2shard.find(node); + it != _original_node2shard.end()) { // this is an original replica, preserve the shard replica.shard = it->second; _state->add_final_count(replica); @@ -120,8 +145,8 @@ replicas_t allocated_partition::release_new_partition( chunked_vector& added_replicas) { for (const auto& bs : _replicas) { if ( - !_original_node2shard - || !_original_node2shard->contains(bs.node_id)) { + _original_node2shard.has_ever_seen_modifications() + && !_original_node2shard.is_original(bs.node_id)) { added_replicas.push_back(bs); } } @@ -131,14 +156,14 @@ replicas_t allocated_partition::release_new_partition( } bool allocated_partition::has_changes() const { - if (!_original_node2shard) { + if (!_original_node2shard.has_ever_seen_modifications()) { return false; } - if (_replicas.size() != _original_node2shard->size()) { + if (_replicas.size() != _original_node2shard.size()) { return true; } for (const auto& bs : _replicas) { - if (!_original_node2shard->contains(bs.node_id)) { + if (!_original_node2shard.is_original(bs.node_id)) { return true; } } @@ -146,8 +171,8 @@ bool allocated_partition::has_changes() const { } bool allocated_partition::is_original(model::node_id node) const { - if (_original_node2shard) { - return _original_node2shard->contains(node); + if (_original_node2shard.has_ever_seen_modifications()) { + return _original_node2shard.is_original(node); } return std::find_if( _replicas.begin(), @@ -162,7 +187,7 @@ errc allocated_partition::try_revert(const reallocation_step& step) { "allocation_state was concurrently replaced"); } - if (!_original_node2shard) { + if (!_original_node2shard.has_ever_seen_modifications()) { return errc::no_update_in_progress; } @@ -184,13 +209,13 @@ errc allocated_partition::try_revert(const reallocation_step& step) { } _state->remove_final_count(step.current()); - if (!_original_node2shard->contains(step.current().node_id)) { + if (!_original_node2shard.is_original(step.current().node_id)) { _state->remove_allocation(step.current()); } if (step.previous()) { _state->add_final_count(*step.previous()); - if (!_original_node2shard->contains(step.previous()->node_id)) { + if (!_original_node2shard.is_original(step.previous()->node_id)) { _state->add_allocation(*step.previous()); } } @@ -201,27 +226,26 @@ errc allocated_partition::try_revert(const reallocation_step& step) { allocated_partition::~allocated_partition() { oncore_debug_verify(_oncore); - if (!_original_node2shard || !_state) { + if (!_original_node2shard.has_ever_seen_modifications() || !_state) { // no new allocations took place or object was moved from return; } for (const auto& bs : _replicas) { - auto orig_it = _original_node2shard->find(bs.node_id); - if (orig_it == _original_node2shard->end()) { - // new replica + if (!_original_node2shard.is_original(bs.node_id)) { _state->remove_allocation(bs); _state->remove_final_count(bs); - } else { - // original replica that didn't change, erase from the map in - // preparation for the loop below - _original_node2shard->erase(orig_it); } } - for (const auto& kv : *_original_node2shard) { - model::broker_shard bs{kv.first, kv.second}; - _state->add_final_count(bs); + // Restore the final count of originals that were moved away. + for (const auto& kv : _original_node2shard.get()) { + const auto node = kv.first; + const bool still_a_replica = std::ranges::any_of( + _replicas, [node](const auto& bs) { return bs.node_id == node; }); + if (!still_a_replica) { + _state->add_final_count(model::broker_shard{node, kv.second}); + } } } fmt::iterator partition_constraints::format_to(fmt::iterator it) const { diff --git a/src/v/cluster/scheduling/types.h b/src/v/cluster/scheduling/types.h index 0658c99aabaeb..d7ebdc7de85c6 100644 --- a/src/v/cluster/scheduling/types.h +++ b/src/v/cluster/scheduling/types.h @@ -26,6 +26,7 @@ #include #include +#include namespace cluster { class allocation_node; @@ -243,6 +244,7 @@ class reallocation_step { private: friend class partition_allocator; + friend class allocated_partition_test; reallocation_step( model::broker_shard current, std::optional previous) : _current(current) @@ -275,6 +277,7 @@ class allocated_partition { private: friend class partition_allocator; + friend class allocated_partition_test; // construct an object from an original assignment allocated_partition( @@ -296,8 +299,38 @@ class allocated_partition { private: model::ntp _ntp; replicas_t _replicas; - std::optional> - _original_node2shard; + + class original_node2shard { + public: + bool has_ever_seen_modifications() const; + void capture(const replicas_t& replicas); + void reset(); + bool is_original(model::node_id node) const; + size_t size() const; + + template + auto find(this Self&& self, model::node_id node) { + vassert(self._snapshot, "original placement snapshot not captured"); + return std::forward(self)._snapshot->find(node); + } + + template + auto end(this Self&& self) { + vassert(self._snapshot, "original placement snapshot not captured"); + return std::forward(self)._snapshot->end(); + } + + template + auto&& get(this Self&& self) { + vassert(self._snapshot, "original placement snapshot not captured"); + return *std::forward(self)._snapshot; + } + + private: + std::optional> _snapshot; + }; + original_node2shard _original_node2shard; + ss::weak_ptr _state; // oncore checker to ensure destruction happens on the same core [[no_unique_address]] oncore _oncore; diff --git a/src/v/cluster/tests/topic_updates_dispatcher_test.cc b/src/v/cluster/tests/topic_updates_dispatcher_test.cc index d565d9dc8b786..78134c042b563 100644 --- a/src/v/cluster/tests/topic_updates_dispatcher_test.cc +++ b/src/v/cluster/tests/topic_updates_dispatcher_test.cc @@ -75,6 +75,53 @@ constexpr uint64_t max_cluster_capacity() { + node_initial_capacity(4); } +// Regression test for a spurious double free which occurred when a no-op +// reallocation was requested (requested replication factor == existing replica +// count). +FIXTURE_TEST( + rf_noop_allocate_spurious_dealloc, topic_table_updates_dispatcher_fixture) { + const auto& alloc_nodes = allocator.local().state().allocation_nodes(); + auto total_allocated = [&]() { + size_t t = 0; + for (const auto& [id, n] : alloc_nodes) { + t += n->allocated_partitions(); + } + return t; + }; + + // Create an RF=3 topic with a single partition: 3 replicas allocated. + auto create = make_create_topic_cmd("ct", 1, 3); + dispatch_command(create); + auto tp_ns = make_tp_ns("ct"); + BOOST_REQUIRE_EQUAL(total_allocated(), 3); + + // Ask the allocator to (re)allocate the same partition at the SAME rf it + // already has (num_new_replicas == 0), exactly as + // increase_replication_factor does for a partition that is already at the + // target RF. + { + auto md = table.local().get_topic_metadata(tp_ns); + BOOST_REQUIRE(md.has_value()); + cluster::allocation_request req(tp_ns); + for (const auto& [id, p_as] : md->get_assignments()) { + req.partitions.emplace_back(p_as, uint16_t{3}); + } + auto res = allocator.local().allocate(std::move(req)).get(); + BOOST_REQUIRE(res.has_value()); + logger.info( + "allocate(rf==existing) held: total allocated = {}", + total_allocated()); + // units destroyed here -> ~allocation_units releases _added_replicas + } + + // The partition's replica set never changed, so all 3 allocations must + // still be present. With the bug, ~allocation_units spuriously removed + // them and this is 0. + logger.info( + "after units released: total allocated = {}", total_allocated()); + BOOST_CHECK_EQUAL(total_allocated(), 3); +} + FIXTURE_TEST( test_dispatching_happy_path_create, topic_table_updates_dispatcher_fixture) { create_topics();