Include logger name in FnApiLogRecordHandler log entries

This change adds the logger name (record.name) to the custom_data field
of LogEntry protos in FnApiLogRecordHandler. This is useful for filtering
and debugging logs in complex pipelines.

Fixes #37146

Author: userzhy

CHANGES.md

@@ -74,6 +74,7 @@
 
 * Support configuring Firestore database on ReadFn transforms (Java) ([#36904](https://github.com/apache/beam/issues/36904)).
 * (Python) Inference args are now allowed in most model handlers, except where they are explicitly/intentionally disallowed ([#37093](https://github.com/apache/beam/issues/37093)).
+* Include logger name in FnApiLogRecordHandler log entries for filtering and debugging (Python) ([#37146](https://github.com/apache/beam/issues/37146)).
 
 ## Breaking Changes
 

sdks/python/apache_beam/runners/worker/log_handler.py

@@ -144,6 +144,10 @@ def emit(self, record: logging.LogRecord) -> None:
           current_state.name_context.transform_id):
         log_entry.transform_id = current_state.name_context.transform_id
 
+    # Include the logger name in custom_data for filtering and debugging.
+    if record.name:
+      log_entry.custom_data.fields['logger'].string_value = record.name
+
     try:
       self._log_entry_queue.put(log_entry, block=False)
     except queue.Full:

sdks/python/apache_beam/runners/worker/log_handler_test.py

@@ -233,6 +233,15 @@ def test_context(self):
     finally:
       statesampler.set_current_tracker(None)
 
+  def test_logger_name_in_custom_data(self):
+    """Tests that logger name is included in custom_data."""
+    _LOGGER.info('test message')
+    self.fn_log_handler.close()
+
+    log_entry = self.test_logging_service.log_records_received[0].log_entries[0]
+    self.assertEqual(
+        log_entry.custom_data.fields['logger'].string_value, __name__)
+
   def test_extracts_transform_id_during_exceptions(self):
     """Tests that transform ids are captured during user code exceptions."""
     descriptor = beam_fn_api_pb2.ProcessBundleDescriptor()