[v24.2.x] bazel: support latest protobuf version

MODULE.bazel

@@ -32,7 +32,7 @@ bazel_dep(name = "googletest", version = "1.14.0.bcr.1")
 bazel_dep(name = "liburing", version = "2.5")
 bazel_dep(name = "lz4", version = "1.9.4")
 bazel_dep(name = "platforms", version = "0.0.10")
-bazel_dep(name = "protobuf", version = "26.0")
+bazel_dep(name = "protobuf", version = "27.3")
 bazel_dep(name = "re2", version = "2023-09-01")
 bazel_dep(name = "rules_foreign_cc", version = "0.10.1")
 bazel_dep(name = "rules_proto", version = "6.0.2")

src/v/pandaproxy/schema_registry/protobuf.cc

@@ -22,6 +22,7 @@
 #include "utils/base64.h"
 
 #include <seastar/core/coroutine.hh>
+#include <seastar/core/sstring.hh>
 
 #include <absl/container/flat_hash_set.h>
 #include <boost/algorithm/string/trim.hpp>
@@ -143,12 +144,23 @@ class io_error_collector final : public pb::io::ErrorCollector {
     };
 
 public:
+#if PROTOBUF_VERSION < 5027000
     void AddError(int line, int column, const std::string& message) final {
         _errors.emplace_back(err{level::error, line, column, message});
     }
     void AddWarning(int line, int column, const std::string& message) final {
         _errors.emplace_back(err{level::warn, line, column, message});
     }
+#else
+    void RecordError(int line, int column, std::string_view message) final {
+        _errors.emplace_back(
+          err{level::error, line, column, ss::sstring{message}});
+    }
+    void RecordWarning(int line, int column, std::string_view message) final {
+        _errors.emplace_back(
+          err{level::warn, line, column, ss::sstring{message}});
+    }
+#endif
 
     error_info error() const;
 
@@ -160,25 +172,72 @@ class io_error_collector final : public pb::io::ErrorCollector {
 
 class dp_error_collector final : public pb::DescriptorPool::ErrorCollector {
 public:
+#if PROTOBUF_VERSION < 5027000
     void AddError(
       const std::string& filename,
       const std::string& element_name,
       const pb::Message* descriptor,
       ErrorLocation location,
       const std::string& message) final {
         _errors.emplace_back(err{
-          level::error, filename, element_name, descriptor, location, message});
+          level::error,
+          ss::sstring{filename},
+          ss::sstring{element_name},
+          descriptor,
+          location,
+          ss::sstring {
+              message
+          }});
     }
+
     void AddWarning(
       const std::string& filename,
       const std::string& element_name,
       const pb::Message* descriptor,
       ErrorLocation location,
       const std::string& message) final {
         _errors.emplace_back(err{
-          level::warn, filename, element_name, descriptor, location, message});
+          level::warn,
+          ss::sstring{filename},
+          ss::sstring{element_name},
+          descriptor,
+          location,
+          ss::sstring {
+              message
+          }});
+    }
+#else
+    void RecordError(
+      std::string_view filename,
+      std::string_view element_name,
+      const pb::Message* descriptor,
+      ErrorLocation location,
+      std::string_view message) final {
+        _errors.emplace_back(err{
+          level::error,
+          ss::sstring{filename},
+          ss::sstring{element_name},
+          descriptor,
+          location,
+          ss::sstring{message}});
     }
 
+    void RecordWarning(
+      std::string_view filename,
+      std::string_view element_name,
+      const pb::Message* descriptor,
+      ErrorLocation location,
+      std::string_view message) final {
+        _errors.emplace_back(err{
+          level::warn,
+          ss::sstring{filename},
+          ss::sstring{element_name},
+          descriptor,
+          location,
+          ss::sstring{message}});
+    }
+#endif
+
     error_info error() const;
 
 private:
@@ -188,11 +247,11 @@ class dp_error_collector final : public pb::DescriptorPool::ErrorCollector {
     };
     struct err {
         level lvl;
-        std::string filename;
-        std::string element_name;
+        ss::sstring filename;
+        ss::sstring element_name;
         const pb::Message* descriptor;
         ErrorLocation location;
-        std::string message;
+        ss::sstring message;
     };
     friend struct fmt::formatter<err>;
 
@@ -243,7 +302,12 @@ class parser {
                 throw as_exception(error_collector.error());
             }
         }
-        _fdp.set_name(schema.sub()());
+        const auto& sub = schema.sub()();
+#if PROTOBUF_VERSION < 5027000
+        _fdp.set_name(sub.data(), sub.size());
+#else
+        _fdp.set_name(std::string_view(sub));
+#endif
         return _fdp;
     }
 
@@ -275,8 +339,8 @@ build_file(pb::DescriptorPool& dp, const pb::FileDescriptorProto& fdp) {
 
 ///\brief Build a FileDescriptor and import references from the store.
 ///
-/// Recursively import references into the DescriptorPool, building the files
-/// on stack unwind.
+/// Recursively import references into the DescriptorPool, building the
+/// files on stack unwind.
 ss::future<const pb::FileDescriptor*> build_file_with_refs(
   pb::DescriptorPool& dp, sharded_store& store, canonical_schema schema) {
     for (const auto& ref : schema.def().refs()) {
@@ -295,7 +359,8 @@ ss::future<const pb::FileDescriptor*> build_file_with_refs(
     co_return build_file(dp, p.parse(schema));
 }
 
-///\brief Import a schema in the DescriptorPool and return the FileDescriptor.
+///\brief Import a schema in the DescriptorPool and return the
+/// FileDescriptor.
 ss::future<const pb::FileDescriptor*> import_schema(
   pb::DescriptorPool& dp, sharded_store& store, canonical_schema schema) {
     try {

src/v/redpanda/application.cc

@@ -147,6 +147,7 @@
 #include <seastar/util/defer.hh>
 #include <seastar/util/log.hh>
 
+#include <absl/log/globals.h>
 #include <google/protobuf/stubs/logging.h>
 #include <sys/resource.h>
 #include <sys/utsname.h>
@@ -543,16 +544,21 @@ void application::initialize(
      * Disable the logger for protobuf; some interfaces don't allow a pluggable
      * error collector.
      */
+#if PROTOBUF_VERSION < 5027000
     google::protobuf::SetLogHandler(nullptr);
+#else
+    // Protobuf uses absl logging in the latest version
+    absl::SetMinLogLevel(absl::LogSeverityAtLeast::kInfinity);
+#endif
 
     /*
      * allocate per-core zstd decompression workspace and per-core
-     * async_stream_zstd workspaces. it can be several megabytes in size, so do
-     * it before memory becomes fragmented.
+     * async_stream_zstd workspaces. it can be several megabytes in size, so
+     * do it before memory becomes fragmented.
      */
     ss::smp::invoke_on_all([] {
-        // TODO: remove this when stream_zstd is replaced with async_stream_zstd
-        // in v/kafka
+        // TODO: remove this when stream_zstd is replaced with
+        // async_stream_zstd in v/kafka
         compression::stream_zstd::init_workspace(
           config::shard_local_cfg().zstd_decompress_workspace_bytes());