Merge pull request #24563 from vbotbuildovich/backport-pr-24555-v24.3.x-909

[v24.3.x] dl/coordinator: add debug logging in coordinator

Author: andrwng

src/v/datalake/coordinator/iceberg_file_committer.cc

@@ -99,17 +99,24 @@ ss::future<
   checked<chunked_vector<mark_files_committed_update>, file_committer::errc>>
 iceberg_file_committer::commit_topic_files_to_catalog(
   model::topic topic, const topics_state& state) const {
+    vlog(datalake_log.debug, "Beginning commit for topic {}", topic);
     auto tp_it = state.topic_to_state.find(topic);
     if (
       tp_it == state.topic_to_state.end()
       || !tp_it->second.has_pending_entries()) {
+        vlog(datalake_log.debug, "Topic {} has no pending entries", topic);
         co_return chunked_vector<mark_files_committed_update>{};
     }
     auto topic_revision = tp_it->second.revision;
 
     auto table_id = table_id_for_topic(topic);
     auto table_res = co_await load_table(table_id);
     if (table_res.has_error()) {
+        vlog(
+          datalake_log.warn,
+          "Error loading table {} for committing from topic {}",
+          table_id,
+          topic);
         co_return table_res.error();
     }
     auto& table = table_res.value();
@@ -130,6 +137,13 @@ iceberg_file_committer::commit_topic_files_to_catalog(
     if (
       tp_it == state.topic_to_state.end()
       || tp_it->second.revision != topic_revision) {
+        vlog(
+          datalake_log.debug,
+          "Commit returning early, topic {} state is missing or revision has "
+          "changed: current {} vs expected {}",
+          topic,
+          tp_it->second.revision,
+          topic_revision);
         co_return chunked_vector<mark_files_committed_update>{};
     }
 
@@ -148,6 +162,16 @@ iceberg_file_committer::commit_topic_files_to_catalog(
                 // replicate the fact that it was committed previously, but
                 // don't construct a data_file to send to Iceberg as it is
                 // already committed.
+                vlog(
+                  datalake_log.debug,
+                  "Skipping entry for topic {} revision {} added at "
+                  "coordinator offset {} because table {} has data including "
+                  "coordinator offset {}",
+                  topic,
+                  topic_revision,
+                  e.added_pending_at,
+                  table_id,
+                  *iceberg_commit_meta_opt);
                 continue;
             }
             new_committed_offset = std::max(
@@ -168,6 +192,11 @@ iceberg_file_committer::commit_topic_files_to_catalog(
         }
     }
     if (pending_commits.empty()) {
+        vlog(
+          datalake_log.debug,
+          "No new data to mark committed for topic {} revision {}",
+          topic,
+          topic_revision);
         co_return chunked_vector<mark_files_committed_update>{};
     }
     chunked_vector<mark_files_committed_update> updates;
@@ -187,6 +216,12 @@ iceberg_file_committer::commit_topic_files_to_catalog(
     }
     if (icb_files.empty()) {
         // All files are deduplicated.
+        vlog(
+          datalake_log.debug,
+          "All committed files were deduplicated for topic {} revision {}, "
+          "returning without updating Iceberg catalog",
+          topic,
+          topic_revision);
         co_return updates;
     }
     vassert(
@@ -195,6 +230,11 @@ iceberg_file_committer::commit_topic_files_to_catalog(
     const auto commit_meta = commit_offset_metadata{
       .offset = *new_committed_offset,
     };
+    vlog(
+      datalake_log.debug,
+      "Adding {} files to Iceberg table {}",
+      icb_files.size(),
+      table_id);
     iceberg::transaction txn(std::move(table));
     auto icb_append_res = co_await txn.merge_append(
       io_,

src/v/datalake/coordinator/state_machine.cc

@@ -110,17 +110,15 @@ ss::future<> coordinator_stm::do_apply(const model::record_batch& b) {
         // TODO: make updates a variant so we can share code more easily?
         case update_key::add_files: {
             auto update = co_await serde::read_async<add_files_update>(val_p);
-            vlog(
-              _log.debug, "Applying {} from offset {}: {}", key, o, update.tp);
+            vlog(_log.debug, "Applying {} from offset {}: {}", key, o, update);
             auto res = update.apply(state_, o);
             maybe_log_update_error(_log, key, o, res);
             continue;
         }
         case update_key::mark_files_committed: {
             auto update
               = co_await serde::read_async<mark_files_committed_update>(val_p);
-            vlog(
-              _log.debug, "Applying {} from offset {}: {}", key, o, update.tp);
+            vlog(_log.debug, "Applying {} from offset {}: {}", key, o, update);
             auto res = update.apply(state_);
             maybe_log_update_error(_log, key, o, res);
             continue;

src/v/datalake/coordinator/state_update.cc

@@ -284,7 +284,34 @@ topic_lifecycle_update::apply(topics_state& state) {
     return true;
 }
 
-std::ostream& operator<<(std::ostream& o, topic_lifecycle_update u) {
+std::ostream& operator<<(std::ostream& o, const add_files_update& u) {
+    fmt::print(o, "{{tp: {}, revision: {}, entries: [", u.tp, u.topic_revision);
+    static constexpr size_t max_to_log = 6;
+    static constexpr size_t halved = max_to_log / 2;
+    const auto& e = u.entries;
+    if (e.size() <= max_to_log) {
+        fmt::print(o, "{}", fmt::join(e, ", "));
+    } else {
+        fmt::print(o, "{}", fmt::join(e.begin(), e.begin() + halved, ", "));
+        o << "...";
+        fmt::print(o, "{}", fmt::join(e.end() - halved, e.end(), ", "));
+    }
+    fmt::print(o, "] ({} entries)}}", e.size());
+    return o;
+}
+
+std::ostream&
+operator<<(std::ostream& o, const mark_files_committed_update& u) {
+    fmt::print(
+      o,
+      "{{tp: {}, revision: {}, new_committed: {}}}",
+      u.tp,
+      u.topic_revision,
+      u.new_committed);
+    return o;
+}
+
+std::ostream& operator<<(std::ostream& o, const topic_lifecycle_update& u) {
     fmt::print(
       o,
       "{{topic: {}, revision: {}, new_state: {}}}",

src/v/datalake/coordinator/state_update.h

@@ -45,6 +45,7 @@ struct add_files_update
     checked<std::nullopt_t, stm_update_error> can_apply(const topics_state&);
     checked<std::nullopt_t, stm_update_error>
     apply(topics_state&, model::offset);
+    friend std::ostream& operator<<(std::ostream&, const add_files_update&);
 
     model::topic_partition tp;
     model::revision_id topic_revision;
@@ -68,6 +69,8 @@ struct mark_files_committed_update
 
     checked<std::nullopt_t, stm_update_error> can_apply(const topics_state&);
     checked<std::nullopt_t, stm_update_error> apply(topics_state&);
+    friend std::ostream&
+    operator<<(std::ostream&, const mark_files_committed_update&);
 
     model::topic_partition tp;
     model::revision_id topic_revision;
@@ -90,7 +93,8 @@ struct topic_lifecycle_update
     checked<bool, stm_update_error> can_apply(const topics_state&);
     checked<bool, stm_update_error> apply(topics_state&);
 
-    friend std::ostream& operator<<(std::ostream&, topic_lifecycle_update);
+    friend std::ostream&
+    operator<<(std::ostream&, const topic_lifecycle_update&);
 
     model::topic topic;
     model::revision_id revision;