datalake: relax translation_stm::max_collectible_offset() value (and add compaction_test.py)

src/v/datalake/translation/BUILD

@@ -13,6 +13,24 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "utils",
+    srcs = [
+        "utils.cc",
+    ],
+    hdrs = [
+        "utils.h",
+    ],
+    include_prefix = "datalake/translation",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//src/v/model",
+        "//src/v/serde",
+        "//src/v/storage",
+        "@seastar",
+    ],
+)
+
 redpanda_cc_library(
     name = "stm",
     srcs = [
@@ -25,6 +43,7 @@ redpanda_cc_library(
     visibility = ["//visibility:public"],
     deps = [
         ":model",
+        ":utils",
         "//src/v/cluster:state_machine_registry",
         "//src/v/datalake:logger",
         "//src/v/raft",
@@ -45,6 +64,7 @@ redpanda_cc_library(
     deps = [
         ":model",
         ":stm",
+        ":utils",
         "//src/v/base",
         "//src/v/cluster",
         "//src/v/datalake:cloud_data_io",

src/v/datalake/translation/CMakeLists.txt

@@ -3,6 +3,7 @@ v_cc_library(
     SRCS
         partition_translator.cc
         state_machine.cc
+	utils.cc
     DEPS
         v::cluster
         v::datalake_writer

src/v/datalake/translation/state_machine.cc

@@ -12,6 +12,7 @@
 
 #include "datalake/logger.h"
 #include "datalake/translation/types.h"
+#include "datalake/translation/utils.h"
 
 namespace {
 raft::replicate_options make_replicate_options() {
@@ -116,8 +117,9 @@ model::offset translation_stm::max_collectible_offset() {
     if (_highest_translated_offset == kafka::offset{}) {
         return model::offset{};
     }
-    return _raft->log()->to_log_offset(
-      kafka::offset_cast(_highest_translated_offset));
+
+    return highest_log_offset_below_next(
+      _raft->log(), _highest_translated_offset);
 }
 
 ss::future<raft::local_snapshot_applied> translation_stm::apply_local_snapshot(

src/v/datalake/translation/tests/BUILD

@@ -9,6 +9,7 @@ redpanda_cc_gtest(
     cpu = 1,
     deps = [
         "//src/v/datalake/translation:stm",
+        "//src/v/datalake/translation:utils",
         "//src/v/raft/tests:stm_test_fixture",
         "//src/v/random:generators",
         "//src/v/storage",

src/v/datalake/translation/tests/CMakeLists.txt

@@ -12,3 +12,14 @@ rp_test(
   LABELS datalake
   ARGS "-- -c 1 -m 2G"
 )
+
+rp_test(
+  UNIT_TEST
+  GTEST
+  TIMEOUT 2000
+  BINARY_NAME translated_log_offset_tests
+  SOURCES translated_log_offset_test.cc
+  LIBRARIES  v::datalake_translation v::features v::raft_fixture v::gtest_main v::storage v::storage_test_utils
+  LABELS datalake
+  ARGS "-- -c 1"
+)

src/v/datalake/translation/tests/state_machine_test.cc

@@ -8,6 +8,7 @@
  * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
  */
 #include "datalake/translation/state_machine.h"
+#include "datalake/translation/utils.h"
 #include "raft/tests/stm_test_fixture.h"
 #include "storage/disk_log_impl.h"
 #include "test_utils/scoped_config.h"
@@ -139,8 +140,9 @@ TEST_F_CORO(translator_stm_fixture, state_machine_ops) {
     model::offset max_collectible_offset{};
     {
         auto log = std::get<0>(node_stms.begin()->second)->raft()->log();
-        max_collectible_offset = log->to_log_offset(
-          kafka::offset_cast(new_translated_offset));
+        max_collectible_offset
+          = datalake::translation::highest_log_offset_below_next(
+            log, new_translated_offset);
     }
     co_await check_max_collectible_offset(max_collectible_offset);
     co_await check_highest_translated_offset(new_translated_offset);