From d44952289b6aa7b228c1a8b57fef9875fabbb507 Mon Sep 17 00:00:00 2001
From: Sanjay Santhanam <51058514+Sanjays2402@users.noreply.github.com>
Date: Sat, 25 Jul 2026 01:15:06 -0700
Subject: [PATCH] fix: use bytes replacement in JobTask.dump for __main__ jobs

luigi.contrib.hadoop.JobTask.dump() rewrote the pickled module name with
d.replace(b"(c__main__", "(c" + module_name), mixing a bytes pattern with
a str replacement. Since pickle.dumps() returns bytes and sys.argv[0] is a
str, this always raised TypeError: a bytes-like object is required, not
'str', so dumping a job defined in __main__ was impossible.

The replacement is now encoded to bytes. Added a regression test that
dumps a __main__-module job and fails with the reported TypeError without
this fix.

Closes #3284
---
 luigi/contrib/hadoop.py     |  2 +-
 test/contrib/hadoop_test.py | 22 ++++++++++++++++++++++
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/luigi/contrib/hadoop.py b/luigi/contrib/hadoop.py
index a712465697..d04093b2e4 100644
--- a/luigi/contrib/hadoop.py
+++ b/luigi/contrib/hadoop.py
@@ -957,7 +957,7 @@ def dump(self, directory=""):
             if self.__module__ == "__main__":
                 d = pickle.dumps(self)
                 module_name = os.path.basename(sys.argv[0]).rsplit(".", 1)[0]
-                d = d.replace(b"(c__main__", "(c" + module_name)
+                d = d.replace(b"(c__main__", b"(c" + module_name.encode("utf-8"))
                 open(file_name, "wb").write(d)
 
             else:
diff --git a/test/contrib/hadoop_test.py b/test/contrib/hadoop_test.py
index 8e1c2fe64d..864a35683c 100644
--- a/test/contrib/hadoop_test.py
+++ b/test/contrib/hadoop_test.py
@@ -18,6 +18,7 @@
 import os
 import sys
 import json
+import tempfile
 import unittest
 
 import luigi
@@ -498,3 +499,24 @@ def test_kill_last_application_on_interrupt(self):
         ]
         subprocess = self._run_and_track_with_interrupt(err_lines)
         subprocess.call.assert_called_once_with(['yarn', 'application', '-kill', application_id])
+
+
+class JobTaskDumpTest(unittest.TestCase):
+    def test_dump_from_main_module(self):
+        """`dump` must not raise TypeError for a __main__ job (issue #3284)."""
+        directory = tempfile.mkdtemp()
+        job = MyStreamingJob(param='x')
+        main_module = sys.modules['__main__']
+        original_module = MyStreamingJob.__module__
+        original_argv0 = sys.argv[0]
+        MyStreamingJob.__module__ = '__main__'
+        setattr(main_module, 'MyStreamingJob', MyStreamingJob)
+        sys.argv[0] = 'my_job_script.py'
+        try:
+            job.dump(directory)
+        finally:
+            MyStreamingJob.__module__ = original_module
+            delattr(main_module, 'MyStreamingJob')
+            sys.argv[0] = original_argv0
+
+        self.assertTrue(os.path.exists(os.path.join(directory, 'job-instance.pickle')))