Merge pull request #13867 from vbotbuildovich/backport-pr-13740-v23.2.x-999

[v23.2.x] cloud_storage: serialize uint64 to topic manifest

Author: piyushredpanda

src/v/cloud_storage/tests/topic_manifest_test.cc

@@ -26,6 +26,9 @@
 #include <boost/test/unit_test.hpp>
 
 #include <chrono>
+#include <limits>
+#include <ratio>
+#include <sstream>
 #include <string>
 #include <string_view>
 #include <system_error>
@@ -112,6 +115,22 @@ static constexpr std::string_view wrong_compaction_strategy = R"json({
     "retention_duration": 36000000000
 })json";
 
+static constexpr std::string_view negative_properties_manifest = R"json({
+    "version": 1,
+    "namespace": "negative-test-namespace",
+    "topic": "full-test-topic",
+    "partition_count": 64,
+    "replication_factor": 6,
+    "revision_id": 1,
+    "compression": "snappy",
+    "cleanup_policy_bitflags": "compact,delete",
+    "compaction_strategy": "offset",
+    "timestamp_type": "LogAppendTime",
+    "segment_size": -1234,
+    "retention_bytes": -42342,
+    "retention_duration": -36000000000
+})json";
+
 inline ss::input_stream<char> make_manifest_stream(std::string_view json) {
     iobuf i;
     i.append(json.data(), json.size());
@@ -224,6 +243,46 @@ SEASTAR_THREAD_TEST_CASE(full_config_update_all_fields_correct) {
       std::chrono::milliseconds(36000000000));
 }
 
+SEASTAR_THREAD_TEST_CASE(topic_manifest_min_serialization) {
+    manifest_topic_configuration min_cfg{cfg};
+    min_cfg.properties.retention_bytes = tristate<size_t>(
+      std::numeric_limits<size_t>::min());
+    min_cfg.properties.retention_duration = tristate<std::chrono::milliseconds>(
+      std::chrono::milliseconds::min());
+    min_cfg.properties.segment_size = std::make_optional(
+      std::numeric_limits<size_t>::min());
+    topic_manifest m(min_cfg, model::initial_revision_id{0});
+    auto [is, size] = m.serialize().get();
+    iobuf buf;
+    auto os = make_iobuf_ref_output_stream(buf);
+    ss::copy(is, os).get();
+
+    auto rstr = make_iobuf_input_stream(std::move(buf));
+    topic_manifest restored;
+    restored.update(std::move(rstr)).get();
+    BOOST_REQUIRE(m == restored);
+}
+
+SEASTAR_THREAD_TEST_CASE(topic_manifest_max_serialization) {
+    manifest_topic_configuration max_cfg{cfg};
+    max_cfg.properties.retention_bytes = tristate<size_t>(
+      std::numeric_limits<size_t>::max());
+    max_cfg.properties.retention_duration = tristate<std::chrono::milliseconds>(
+      std::chrono::milliseconds::max());
+    max_cfg.properties.segment_size = std::make_optional(
+      std::numeric_limits<size_t>::max());
+    topic_manifest m(max_cfg, model::initial_revision_id{0});
+    auto [is, size] = m.serialize().get();
+    iobuf buf;
+    auto os = make_iobuf_ref_output_stream(buf);
+    ss::copy(is, os).get();
+
+    auto rstr = make_iobuf_input_stream(std::move(buf));
+    topic_manifest restored;
+    restored.update(std::move(rstr)).get();
+    BOOST_REQUIRE(m == restored);
+}
+
 SEASTAR_THREAD_TEST_CASE(missing_required_fields_throws) {
     topic_manifest m;
     BOOST_REQUIRE_EXCEPTION(
@@ -294,3 +353,20 @@ SEASTAR_THREAD_TEST_CASE(update_non_empty_manifest) {
 
     BOOST_REQUIRE(m == restored);
 }
+
+SEASTAR_THREAD_TEST_CASE(test_negative_property_manifest) {
+    topic_manifest m(cfg, model::initial_revision_id(0));
+    m.update(make_manifest_stream(negative_properties_manifest)).get();
+    auto tp_cfg = m.get_topic_config();
+    BOOST_REQUIRE(tp_cfg.has_value());
+    BOOST_REQUIRE_EQUAL(64, tp_cfg->partition_count);
+    BOOST_REQUIRE_EQUAL(6, tp_cfg->replication_factor);
+    auto tp_props = tp_cfg->properties;
+    BOOST_REQUIRE(tp_props.retention_duration.has_optional_value());
+    BOOST_REQUIRE_EQUAL(
+      tp_props.retention_duration.value().count(), -36000000000);
+
+    // The usigned types that were passed in negative values shouldn't be set.
+    BOOST_REQUIRE(tp_props.retention_bytes.is_disabled());
+    BOOST_REQUIRE(!tp_props.segment_size.has_value());
+}

src/v/cloud_storage/topic_manifest.cc

@@ -94,6 +94,46 @@ struct topic_manifest_handler
         }
     }
 
+    bool Int(int i) { return Int64(i); }
+
+    bool Int64(int64_t i) {
+        if (i >= 0) {
+            // Should only be called when negative, but doesn't hurt to just
+            // defer to the unsigned variant.
+            return Uint64(i);
+        }
+        switch (_state) {
+        case state::expect_value:
+            if (_key == "version") {
+                _version = i;
+            } else if (_key == "partition_count") {
+                _partition_count = i;
+            } else if (_key == "replication_factor") {
+                _replication_factor = i;
+            } else if (_key == "revision_id") {
+                _revision_id = model::initial_revision_id{i};
+            } else if (_key == "segment_size") {
+                // NOTE: segment size and retention bytes are unsigned, but
+                // older versions of Redpanda could serialize them as negative.
+                // Just leave them empty.
+                _properties.segment_size = std::nullopt;
+            } else if (_key == "retention_bytes") {
+                _properties.retention_bytes = tristate<size_t>{};
+            } else if (_key == "retention_duration") {
+                _properties.retention_duration
+                  = tristate<std::chrono::milliseconds>(
+                    std::chrono::milliseconds(i));
+            } else {
+                return false;
+            }
+            _state = state::expect_key;
+            return true;
+        case state::expect_manifest_start:
+        case state::expect_key:
+            return false;
+        }
+    }
+
     bool Uint(unsigned u) { return Uint64(u); }
 
     bool Uint64(uint64_t u) {
@@ -375,7 +415,7 @@ void topic_manifest::serialize(std::ostream& out) const {
     }
     w.Key("segment_size");
     if (_topic_config->properties.segment_size.has_value()) {
-        w.Int64(*_topic_config->properties.segment_size);
+        w.Uint64(*_topic_config->properties.segment_size);
     } else {
         w.Null();
     }
@@ -388,7 +428,7 @@ void topic_manifest::serialize(std::ostream& out) const {
     if (!_topic_config->properties.retention_bytes.is_disabled()) {
         w.Key("retention_bytes");
         if (_topic_config->properties.retention_bytes.has_optional_value()) {
-            w.Int64(_topic_config->properties.retention_bytes.value());
+            w.Uint64(_topic_config->properties.retention_bytes.value());
         } else {
             w.Null();
         }