cloud_storage: re-adopt topic_manifest

src/v/cloud_storage/CMakeLists.txt

@@ -38,6 +38,7 @@ v_cc_library(
     segment_chunk.cc
     segment_chunk_api.cc
     segment_chunk_data_source.cc
+    topic_manifest.cc
     async_manifest_view.cc
     materialized_manifest_cache.cc
     anomalies_detector.cc
@@ -48,6 +49,7 @@ v_cc_library(
     Seastar::seastar
     v::bytes
     v::http
+    v::cluster_topic_properties
     v::cloud_storage_clients
     v::json
     v::model

src/v/cluster/CMakeLists.txt

@@ -83,6 +83,18 @@ rpcgen(
   INCLUDES ${CMAKE_BINARY_DIR}/src/v
 )
 
+v_cc_library(
+  NAME cluster_topic_properties
+  SRCS
+    remote_topic_properties.cc
+    topic_configuration.cc
+    topic_properties.cc
+  DEPS
+    v::model
+    v::reflection
+    v::serde
+  )
+
 v_cc_library(
   NAME cluster
   HDRS
@@ -215,9 +227,6 @@ v_cc_library(
     shard_placement_table.cc
     shard_balancer.cc
     topic_recovery_validator.cc
-    ../cloud_storage/topic_manifest.cc
-  HDRS
-    ../cloud_storage/topic_manifest.h
   DEPS
     Seastar::seastar
     bootstrap_rpc
@@ -238,6 +247,7 @@ v_cc_library(
     v::security
     v::model
     v::cloud_storage
+    v::cluster_topic_properties
     v::features
     v::version
   )

src/v/cluster/remote_topic_properties.cc

@@ -0,0 +1,43 @@
+// Copyright 2024 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/remote_topic_properties.h"
+
+#include "reflection/adl.h"
+
+namespace cluster {
+
+std::ostream& operator<<(std::ostream& o, const remote_topic_properties& rtps) {
+    fmt::print(
+      o,
+      "{{remote_revision: {} remote_partition_count: {}}}",
+      rtps.remote_revision,
+      rtps.remote_partition_count);
+    return o;
+}
+
+} // namespace cluster
+
+namespace reflection {
+
+void adl<cluster::remote_topic_properties>::to(
+  iobuf& out, cluster::remote_topic_properties&& p) {
+    reflection::serialize(out, p.remote_revision, p.remote_partition_count);
+}
+
+cluster::remote_topic_properties
+adl<cluster::remote_topic_properties>::from(iobuf_parser& parser) {
+    auto remote_revision = reflection::adl<model::initial_revision_id>{}.from(
+      parser);
+    auto remote_partition_count = reflection::adl<int32_t>{}.from(parser);
+
+    return {remote_revision, remote_partition_count};
+}
+
+} // namespace reflection

src/v/cluster/remote_topic_properties.h

@@ -0,0 +1,57 @@
+// Copyright 2024 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
+
+#pragma once
+
+#include "model/adl_serde.h"
+#include "model/fundamental.h"
+#include "reflection/adl.h"
+#include "serde/envelope.h"
+
+namespace cluster {
+
+struct remote_topic_properties
+  : serde::envelope<
+      remote_topic_properties,
+      serde::version<0>,
+      serde::compat_version<0>> {
+    remote_topic_properties() = default;
+    remote_topic_properties(
+      model::initial_revision_id remote_revision,
+      int32_t remote_partition_count)
+      : remote_revision(remote_revision)
+      , remote_partition_count(remote_partition_count) {}
+
+    model::initial_revision_id remote_revision;
+    int32_t remote_partition_count;
+
+    auto serde_fields() {
+        return std::tie(remote_revision, remote_partition_count);
+    }
+
+    friend bool
+    operator==(const remote_topic_properties&, const remote_topic_properties&)
+      = default;
+
+    friend std::ostream&
+    operator<<(std::ostream&, const remote_topic_properties&);
+};
+
+} // namespace cluster
+// namespace cluster
+
+namespace reflection {
+
+template<>
+struct adl<cluster::remote_topic_properties> {
+    void to(iobuf&, cluster::remote_topic_properties&&);
+    cluster::remote_topic_properties from(iobuf_parser&);
+};
+
+} // namespace reflection

src/v/cluster/topic_configuration.cc

@@ -0,0 +1,160 @@
+// Copyright 2024 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/topic_configuration.h"
+
+#include "model/fundamental.h"
+#include "model/metadata.h"
+#include "reflection/adl.h"
+#include "storage/ntp_config.h"
+
+#include <seastar/core/sstring.hh>
+
+namespace cluster {
+
+storage::ntp_config topic_configuration::make_ntp_config(
+  const ss::sstring& work_dir,
+  model::partition_id p_id,
+  model::revision_id rev,
+  model::initial_revision_id init_rev) const {
+    auto has_overrides = properties.has_overrides() || is_internal();
+    std::unique_ptr<storage::ntp_config::default_overrides> overrides = nullptr;
+
+    if (has_overrides) {
+        overrides = std::make_unique<storage::ntp_config::default_overrides>(
+          storage::ntp_config::default_overrides{
+            .cleanup_policy_bitflags = properties.cleanup_policy_bitflags,
+            .compaction_strategy = properties.compaction_strategy,
+            .segment_size = properties.segment_size,
+            .retention_bytes = properties.retention_bytes,
+            .retention_time = properties.retention_duration,
+            // we disable cache for internal topics as they are read only once
+            // during bootstrap.
+            .cache_enabled = storage::with_cache(!is_internal()),
+            .recovery_enabled = storage::topic_recovery_enabled(
+              properties.recovery ? *properties.recovery : false),
+            .shadow_indexing_mode = properties.shadow_indexing,
+            .read_replica = properties.read_replica,
+            .retention_local_target_bytes
+            = properties.retention_local_target_bytes,
+            .retention_local_target_ms = properties.retention_local_target_ms,
+            .remote_delete = properties.remote_delete,
+            .segment_ms = properties.segment_ms,
+            .initial_retention_local_target_bytes
+            = properties.initial_retention_local_target_bytes,
+            .initial_retention_local_target_ms
+            = properties.initial_retention_local_target_ms,
+            .write_caching = properties.write_caching,
+            .flush_ms = properties.flush_ms,
+            .flush_bytes = properties.flush_bytes,
+          });
+    }
+    return {
+      model::ntp(tp_ns.ns, tp_ns.tp, p_id),
+      work_dir,
+      std::move(overrides),
+      rev,
+      init_rev};
+}
+
+std::ostream& operator<<(std::ostream& o, const topic_configuration& cfg) {
+    fmt::print(
+      o,
+      "{{ topic: {}, partition_count: {}, replication_factor: {}, "
+      "properties: "
+      "{}}}",
+      cfg.tp_ns,
+      cfg.partition_count,
+      cfg.replication_factor,
+      cfg.properties);
+
+    return o;
+}
+
+} // namespace cluster
+
+namespace reflection {
+
+// note: adl serialization doesn't support read replica fields since serde
+// should be used for new versions.
+void adl<cluster::topic_configuration>::to(
+  iobuf& out, cluster::topic_configuration&& t) {
+    int32_t version = -1;
+    reflection::serialize(
+      out,
+      version,
+      t.tp_ns,
+      t.partition_count,
+      t.replication_factor,
+      t.properties.compression,
+      t.properties.cleanup_policy_bitflags,
+      t.properties.compaction_strategy,
+      t.properties.timestamp_type,
+      t.properties.segment_size,
+      t.properties.retention_bytes,
+      t.properties.retention_duration,
+      t.properties.recovery,
+      t.properties.shadow_indexing);
+}
+
+// note: adl deserialization doesn't support read replica fields since serde
+// should be used for new versions.
+cluster::topic_configuration
+adl<cluster::topic_configuration>::from(iobuf_parser& in) {
+    // NOTE: The first field of the topic_configuration is a
+    // model::ns which has length prefix which is always
+    // positive.
+    // We're using negative length value to encode version. So if
+    // the first int32_t value is positive then we're dealing with
+    // the old format. The negative value means that the new format
+    // was used.
+    auto version = adl<int32_t>{}.from(in.peek(4));
+    if (version < 0) {
+        // Consume version from stream
+        in.skip(4);
+        vassert(
+          -1 == version,
+          "topic_configuration version {} is not supported",
+          version);
+    } else {
+        version = 0;
+    }
+    auto ns = model::ns(adl<ss::sstring>{}.from(in));
+    auto topic = model::topic(adl<ss::sstring>{}.from(in));
+    auto partition_count = adl<int32_t>{}.from(in);
+    auto rf = adl<int16_t>{}.from(in);
+
+    auto cfg = cluster::topic_configuration(
+      std::move(ns), std::move(topic), partition_count, rf);
+
+    cfg.properties.compression = adl<std::optional<model::compression>>{}.from(
+      in);
+    cfg.properties.cleanup_policy_bitflags
+      = adl<std::optional<model::cleanup_policy_bitflags>>{}.from(in);
+    cfg.properties.compaction_strategy
+      = adl<std::optional<model::compaction_strategy>>{}.from(in);
+    cfg.properties.timestamp_type
+      = adl<std::optional<model::timestamp_type>>{}.from(in);
+    cfg.properties.segment_size = adl<std::optional<size_t>>{}.from(in);
+    cfg.properties.retention_bytes = adl<tristate<size_t>>{}.from(in);
+    cfg.properties.retention_duration
+      = adl<tristate<std::chrono::milliseconds>>{}.from(in);
+    if (version < 0) {
+        cfg.properties.recovery = adl<std::optional<bool>>{}.from(in);
+        cfg.properties.shadow_indexing
+          = adl<std::optional<model::shadow_indexing_mode>>{}.from(in);
+    }
+
+    // Legacy topics from pre-22.3 get remote delete disabled.
+    cfg.properties.remote_delete = storage::ntp_config::legacy_remote_delete;
+
+    return cfg;
+}
+
+} // namespace reflection