Port redpanda to s390x (big-endian)

MODULE.bazel

@@ -9,6 +9,10 @@ archive_override(
     integrity = "sha256-wQsxOxZh1tluc6AXj+ncsYhBSl03uC+hPhBrU6JkupI=",
     strip_prefix = "rules_boost-f5b0f8c904f2487d8f5a9a956d4388724e627210",
     urls = ["https://github.com/nelhage/rules_boost/archive/f5b0f8c904f2487d8f5a9a956d4388724e627210.tar.gz"],
+    patch_strip = 1,
+    patches = [
+        "//bazel/thirdparty:rules-boost-s390x-context.patch",
+    ],
 )
 
 non_module_boost_repositories = use_extension("@com_github_nelhage_rules_boost//:boost/repositories.bzl", "non_module_dependencies")
@@ -21,6 +25,13 @@ bazel_dep(name = "gazelle", version = "0.45.0")
 bazel_dep(name = "abseil-cpp", version = "20250814.1")
 bazel_dep(name = "bazel_skylib", version = "1.8.2")
 bazel_dep(name = "crc32c", version = "1.1.0")
+single_version_override(
+    module_name = "crc32c",
+    patch_strip = 1,
+    patches = [
+        "//bazel/thirdparty:crc32c-big-endian-config.patch",
+    ],
+)
 bazel_dep(name = "fmt", version = "12.1.0")
 bazel_dep(name = "googletest", version = "1.17.0")
 bazel_dep(name = "lexy", version = "2025.05.0")

bazel/BUILD

@@ -94,3 +94,8 @@ config_setting(
         ":antithesis": "true",
     },
 )
+
+config_setting(
+    name = "s390x",
+    constraint_values = ["@platforms//cpu:s390x"],
+)

bazel/test.bzl

@@ -75,7 +75,6 @@ def _test_options():
     data = [
         "//:ubsan_suppressions",
         "//:lsan_suppressions",
-        "@current_llvm_toolchain//:llvm-symbolizer",
     ]
     env = {
         "BOOST_TEST_LOG_LEVEL": "test_suite",
@@ -84,7 +83,6 @@ def _test_options():
         "BOOST_TEST_REPORT_LEVEL": "no",
         "BOOST_LOGGER": "HRF,test_suite",
         "ASAN_OPTIONS": "disable_coredump=0:abort_on_error=1",
-        "ASAN_SYMBOLIZER_PATH": "$(rootpath @current_llvm_toolchain//:llvm-symbolizer)",
         "LSAN_OPTIONS": "suppressions=$(rootpath //:lsan_suppressions)",
         "UBSAN_OPTIONS": "symbolize=1:print_stacktrace=1:halt_on_error=1:abort_on_error=1:report_error_type=1:suppressions=$(rootpath //:ubsan_suppressions)",
         # see https://redpandadata.atlassian.net/wiki/x/BwDSUw
@@ -93,6 +91,23 @@ def _test_options():
     deps = antithesis_deps()
     return data, env, deps
 
+# The hermetic LLVM toolchain has no s390x distribution, so the
+# llvm-symbolizer target cannot be fetched there. Sanitizers are not used
+# on s390x, so drop the symbolizer data dep and ASAN_SYMBOLIZER_PATH on that
+# arch while keeping them everywhere else.
+_SYMBOLIZER_DATA = select({
+    "//bazel:s390x": [],
+    "//conditions:default": ["@current_llvm_toolchain//:llvm-symbolizer"],
+})
+
+def _symbolizer_env(base_env):
+    return select({
+        "//bazel:s390x": base_env,
+        "//conditions:default": base_env | {
+            "ASAN_SYMBOLIZER_PATH": "$(rootpath @current_llvm_toolchain//:llvm-symbolizer)",
+        },
+    })
+
 def _redpanda_cc_test(
         name,
         timeout,
@@ -171,9 +186,9 @@ def _redpanda_cc_test(
             "layering_check",
         ],
         tags = resource_tags + tags,
-        env = {"RP_FIXTURE_ENV": "1"} | test_env | env,
+        env = _symbolizer_env({"RP_FIXTURE_ENV": "1"} | test_env | env),
         target_compatible_with = target_compatible_with,
-        data = data + test_data,
+        data = data + test_data + _SYMBOLIZER_DATA,
         local_defines = local_defines,
         flaky = flaky,
     )
@@ -215,8 +230,8 @@ def _redpanda_cc_fuzz_test(
         tags = [
             "fuzz",
         ],
-        env = test_env | env,
-        data = data + test_data,
+        env = _symbolizer_env(test_env | env),
+        data = data + test_data + _SYMBOLIZER_DATA,
         linkopts = [
             "-fsanitize=fuzzer",
         ],
@@ -355,8 +370,8 @@ def redpanda_cc_btest_no_seastar(
             "//src/v/test_utils:boost_test_hooks",
             "@boost//:test.so",
         ] + deps + test_deps,
-        data = test_data,
-        env = test_env,
+        data = test_data + _SYMBOLIZER_DATA,
+        env = _symbolizer_env(test_env),
     )
 
 def redpanda_test_cc_library(
@@ -518,7 +533,7 @@ def redpanda_cc_bench(
         main = "bench_wrapper.py",
         tags = resource_tags + tags,
         srcs = ["//bazel:bench_wrapper"],
-        env = test_env | env,
+        env = _symbolizer_env(test_env | env),
         args = test_args,
-        data = [":" + binary_name] + data + test_data,
+        data = [":" + binary_name] + data + test_data + _SYMBOLIZER_DATA,
     )

bazel/thirdparty/crc32c-big-endian-config.patch

@@ -0,0 +1,19 @@
+Set BYTE_ORDER_BIG_ENDIAN from the compiler's byte-order macros instead of
+hardcoding it to 0. The crc32c portable implementation reads input words via
+ReadUint32LE, whose big-endian branch is only compiled when
+BYTE_ORDER_BIG_ENDIAN is non-zero. Hardcoding 0 makes the LE memcpy path run on
+big-endian hosts (e.g. s390x), producing incorrect CRC32C values and breaking
+Kafka record-batch validation. Deriving the value from __BYTE_ORDER__ keeps the
+config correct on every target arch.
+
+--- a/BUILD.bazel
++++ b/BUILD.bazel
+@@ -76,7 +76,7 @@
+     outs = ["crc32c/crc32c_config.h"],
+     cmd = """
+ sed -e 's/#cmakedefine01/#define/' \
+-    -e 's/ BYTE_ORDER_BIG_ENDIAN/ BYTE_ORDER_BIG_ENDIAN 0/' \
++    -e 's/ BYTE_ORDER_BIG_ENDIAN/ BYTE_ORDER_BIG_ENDIAN (__BYTE_ORDER__ == __ORDER_BIG_ENDIAN__)/' \
+     -e 's/ HAVE_BUILTIN_PREFETCH/ HAVE_BUILTIN_PREFETCH 0/' \
+     -e 's/ HAVE_MM_PREFETCH/ HAVE_MM_PREFETCH 0/' \
+     -e 's/ HAVE_SSE42/ HAVE_SSE42 1/' \

bazel/thirdparty/hwloc.BUILD

@@ -23,6 +23,13 @@ configure_make(
     configure_options = [
         "--disable-libudev",
 
+        # Don't let autoconf pick up a system libxml2 for optional XML
+        # topology import/export. We don't use it, and an auto-detected
+        # libxml2 leaks undefined xml* symbols into every hwloc consumer
+        # (the dep isn't wired up), breaking links on hosts that happen to
+        # have libxml2 headers installed.
+        "--disable-libxml2",
+
         # Disable graphics and the many kinds of display driver discovery
         "--disable-gl",
         "--disable-opencl",

bazel/thirdparty/rules-boost-s390x-context.patch

@@ -0,0 +1,29 @@
+--- a/boost.BUILD
++++ b/boost.BUILD
+@@ -24,6 +24,14 @@
+ )
+
+ config_setting(
++    name = "linux_s390x",
++    constraint_values = [
++        "@platforms//os:linux",
++        "@platforms//cpu:s390x",
++    ],
++)
++
++config_setting(
+     name = "linux_aarch64",
+     constraint_values = [
+         "@platforms//os:linux",
+@@ -106,6 +114,11 @@
+         "libs/context/src/asm/make_ppc64_sysv_elf_gas.S",
+         "libs/context/src/asm/ontop_ppc64_sysv_elf_gas.S",
+     ],
++    ":linux_s390x": [
++        "libs/context/src/asm/jump_s390x_sysv_elf_gas.S",
++        "libs/context/src/asm/make_s390x_sysv_elf_gas.S",
++        "libs/context/src/asm/ontop_s390x_sysv_elf_gas.S",
++    ],
+     ":linux_x86_64": [
+         "libs/context/src/asm/jump_x86_64_sysv_elf_gas.S",
+         "libs/context/src/asm/make_x86_64_sysv_elf_gas.S",

src/v/bytes/ioarray.cc

@@ -269,7 +269,9 @@ uint32_t ioarray::read_fixed32(size_t i) const {
     v |= read(++i) << 8u;
     v |= read(++i) << 16u;
     v |= read(++i) << 24u;
-    return ss::le_to_cpu(v);
+    // The shifts above already assemble the bytes into host byte order
+    // (least-significant byte first), so no le_to_cpu is needed here.
+    return v;
 }
 
 void ioarray::trim_back(size_t n) {

src/v/kafka/protocol/tests/batch_reader_test.cc

@@ -123,8 +123,10 @@ SEASTAR_THREAD_TEST_CASE(consumer_records_consume_batch) {
 
 SEASTAR_THREAD_TEST_CASE(consumer_records_consume_batch_fail_magic) {
     auto ctx = make_context(base_offset, few_batches);
-    corrupt_offset<int32_t>(
-      ctx.record_set, mag_offset, [](int32_t& t) { --t; });
+    // The magic is a single byte; corrupt it as int8 so exactly that byte
+    // changes. Corrupting a 4-byte int here only alters the magic byte on
+    // little-endian hosts (where it is the least-significant byte).
+    corrupt_offset<int8_t>(ctx.record_set, mag_offset, [](int8_t& t) { --t; });
 
     auto crs = kafka::batch_reader(std::move(ctx.record_set));
 
@@ -209,8 +211,10 @@ SEASTAR_THREAD_TEST_CASE(batch_reader_record_batch_reader_impl_fail_lod) {
 
 SEASTAR_THREAD_TEST_CASE(batch_reader_record_batch_reader_impl_fail_magic) {
     auto ctx = make_context(base_offset, few_batches);
-    corrupt_offset<int32_t>(
-      ctx.record_set, mag_offset, [](int32_t& t) { --t; });
+    // The magic is a single byte; corrupt it as int8 so exactly that byte
+    // changes. Corrupting a 4-byte int here only alters the magic byte on
+    // little-endian hosts (where it is the least-significant byte).
+    corrupt_offset<int8_t>(ctx.record_set, mag_offset, [](int8_t& t) { --t; });
 
     auto rdr = model::make_record_batch_reader<kafka::batch_reader>(
       std::move(ctx.record_set));

src/v/reflection/adl.h

@@ -83,10 +83,10 @@ struct adl {
             }
             return adl<value_type>{}.from(in);
         } else if constexpr (is_sstring) {
-            return in.read_string(in.template consume_type<int32_t>());
+            return in.read_string(adl<int32_t>{}.from(in));
         } else if constexpr (is_vector) {
             using value_type = typename type::value_type;
-            int32_t n = in.template consume_type<int32_t>();
+            int32_t n = adl<int32_t>{}.from(in);
             std::vector<value_type> ret;
             ret.reserve(n);
             while (n-- > 0) {
@@ -95,30 +95,30 @@ struct adl {
             return ret;
         } else if constexpr (is_chunked_vector || is_chunked_fifo) {
             using value_type = typename type::value_type;
-            int32_t n = in.template consume_type<int32_t>();
+            int32_t n = adl<int32_t>{}.from(in);
             type ret;
             while (n-- > 0) {
                 ret.push_back(adl<value_type>{}.from(in));
             }
             return ret;
         } else if constexpr (is_btree_set) {
             using value_type = typename type::value_type;
-            int32_t n = in.template consume_type<int32_t>();
+            int32_t n = adl<int32_t>{}.from(in);
             absl::btree_set<value_type> ret;
             while (n-- > 0) {
                 ret.insert(adl<value_type>{}.from(in));
             }
             return ret;
         } else if constexpr (is_circular_buffer) {
             using value_type = typename type::value_type;
-            int32_t n = in.template consume_type<int32_t>();
+            int32_t n = adl<int32_t>{}.from(in);
             ss::circular_buffer<value_type> ret;
             while (n-- > 0) {
                 ret.push_back(adl<value_type>{}.from(in));
             }
             return ret;
         } else if constexpr (is_iobuf) {
-            return in.share(in.template consume_type<int32_t>());
+            return in.share(adl<int32_t>{}.from(in));
         } else if constexpr (is_enum) {
             using e_type = std::underlying_type_t<type>;
             return static_cast<type>(adl<e_type>{}.from(in));

src/v/serde/parquet/encoding.cc

@@ -45,9 +45,17 @@ template<typename value_type>
 void numeric_plain_encoder<value_type>::add_value(value_type v) {
     if constexpr (std::is_integral_v<decltype(v.val)>) {
         v.val = ss::cpu_to_le(v.val);
+        // NOLINTNEXTLINE(*reinterpret-cast*)
+        buf.append(reinterpret_cast<const uint8_t*>(&v.val), sizeof(v.val));
+    } else {
+        // PLAIN encoding stores FLOAT/DOUBLE as little-endian IEEE bytes.
+        // cpu_to_le only accepts integrals, so byteswap the bit pattern.
+        using bits_type
+          = std::conditional_t<sizeof(v.val) == 4, uint32_t, uint64_t>;
+        auto bits = ss::cpu_to_le(std::bit_cast<bits_type>(v.val));
+        // NOLINTNEXTLINE(*reinterpret-cast*)
+        buf.append(reinterpret_cast<const uint8_t*>(&bits), sizeof(bits));
     }
-    // NOLINTNEXTLINE(*reinterpret-cast*)
-    buf.append(reinterpret_cast<const uint8_t*>(&v.val), sizeof(v.val));
 }
 
 template<typename value_type>