We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
HdfsTarget is not workging in luigi > 2.6.1
This code runs correct under luigi <= 2.6.1:

```python
from luigi.contrib import hdfs
import luigi.format

# just to be sure that realy the right version is used...
import pkg_resources
print "luigi ==", pkg_resources.get_distribution("luigi").version
print "snakebite ==", pkg_resources.get_distribution("snakebite").version

destination_file = '/tmp/test/file.gz'

target = hdfs.HdfsTarget(path=destination_file, format=luigi.format.Gzip)

if target.exists():
    target.remove(skip_trash=False)

fsobj = target.open('w')
fsobj.write('lol3\n')
fsobj.close()
```

with luigi 2.6.2 or 2.7.0 it breaks:
```python
luigi == 2.7.0
snakebite == 2.11.0
Traceback (most recent call last):
  File "/opt/tests/hdfs_target.py", line 18, in <module>
    fsobj.close()
  File "/opt/python-2.7.10/lib/python2.7/site-packages/luigi/format.py", line 224, in close
    self._output_pipe.close()
  File "/opt/python-2.7.10/lib/python2.7/site-packages/luigi/contrib/hdfs/format.py", line 51, in close
    remove(self.path)
  File "/opt/python-2.7.10/lib/python2.7/site-packages/luigi/contrib/hdfs/clients.py", line 62, in result
    return getattr(get_autoconfig_client(), method_name)(*args, **kwargs)
  File "/opt/python-2.7.10/lib/python2.7/site-packages/luigi/contrib/hdfs/snakebite_client.py", line 140, in remove
    return list(self.get_bite().delete(self.list_path(path), recurse=recursive))
  File "/opt/python-2.7.10/lib/python2.7/site-packages/snakebite/client.py", line 1540, in wrapped
    yield results.next()
  File "/opt/python-2.7.10/lib/python2.7/site-packages/snakebite/client.py", line 508, in delete
    for item in self._find_items(paths, processor, include_toplevel=True):
  File "/opt/python-2.7.10/lib/python2.7/site-packages/snakebite/client.py", line 1216, in _find_items
    raise FileNotFoundException("`%s': No such file or directory" % path)
snakebite.errors.FileNotFoundException: `/tmp/test/file.gz': No such file or directory
```

client.cfg:
```
[hdfs]
client = snakebite
snakebite_autoconfig = True
tmp_dir: /x/tmp
```

The file is correct in form and content written in "tmp_dir" (/x/tmp/username/test/file.gz-luigitemp-951771388).

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `luigi/contrib/hdfs/format.py`
Content:
```
1 import logging
2 import os
3 
4 import luigi.format
5 from luigi.contrib.hdfs.config import load_hadoop_cmd
6 from luigi.contrib.hdfs import config as hdfs_config
7 from luigi.contrib.hdfs.clients import remove, rename, mkdir, listdir, exists
8 from luigi.contrib.hdfs.error import HDFSCliError
9 
10 logger = logging.getLogger('luigi-interface')
11 
12 
13 class HdfsAtomicWriteError(IOError):
14     pass
15 
16 
17 class HdfsReadPipe(luigi.format.InputPipeProcessWrapper):
18 
19     def __init__(self, path):
20         super(HdfsReadPipe, self).__init__(load_hadoop_cmd() + ['fs', '-cat', path])
21 
22 
23 class HdfsAtomicWritePipe(luigi.format.OutputPipeProcessWrapper):
24     """
25     File like object for writing to HDFS
26 
27     The referenced file is first written to a temporary location and then
28     renamed to final location on close(). If close() isn't called
29     the temporary file will be cleaned up when this object is
30     garbage collected
31 
32     TODO: if this is buggy, change it so it first writes to a
33     local temporary file and then uploads it on completion
34     """
35 
36     def __init__(self, path):
37         self.path = path
38         self.tmppath = hdfs_config.tmppath(self.path)
39         parent_dir = os.path.dirname(self.tmppath)
40         mkdir(parent_dir, parents=True, raise_if_exists=False)
41         super(HdfsAtomicWritePipe, self).__init__(load_hadoop_cmd() + ['fs', '-put', '-', self.tmppath])
42 
43     def abort(self):
44         logger.info("Aborting %s('%s'). Removing temporary file '%s'",
45                     self.__class__.__name__, self.path, self.tmppath)
46         super(HdfsAtomicWritePipe, self).abort()
47         remove(self.tmppath, skip_trash=True)
48 
49     def close(self):
50         super(HdfsAtomicWritePipe, self).close()
51         try:
52             remove(self.path)
53         except HDFSCliError:
54             pass
55         if not all(result['result'] for result in rename(self.tmppath, self.path) or []):
56             raise HdfsAtomicWriteError('Atomic write to {} failed'.format(self.path))
57 
58 
59 class HdfsAtomicWriteDirPipe(luigi.format.OutputPipeProcessWrapper):
60     """
61     Writes a data<data_extension> file to a directory at <path>.
62     """
63 
64     def __init__(self, path, data_extension=""):
65         self.path = path
66         self.tmppath = hdfs_config.tmppath(self.path)
67         self.datapath = self.tmppath + ("/data%s" % data_extension)
68         super(HdfsAtomicWriteDirPipe, self).__init__(load_hadoop_cmd() + ['fs', '-put', '-', self.datapath])
69 
70     def abort(self):
71         logger.info("Aborting %s('%s'). Removing temporary dir '%s'",
72                     self.__class__.__name__, self.path, self.tmppath)
73         super(HdfsAtomicWriteDirPipe, self).abort()
74         remove(self.tmppath, skip_trash=True)
75 
76     def close(self):
77         super(HdfsAtomicWriteDirPipe, self).close()
78         try:
79             if exists(self.path):
80                 remove(self.path)
81         except Exception as ex:
82             if isinstance(ex, HDFSCliError) or ex.args[0].contains("FileNotFoundException"):
83                 pass
84             else:
85                 raise ex
86 
87         # it's unlikely to fail in this way but better safe than sorry
88         if not all(result['result'] for result in rename(self.tmppath, self.path) or []):
89             raise HdfsAtomicWriteError('Atomic write to {} failed'.format(self.path))
90 
91         if os.path.basename(self.tmppath) in map(os.path.basename, listdir(self.path)):
92             remove(self.path)
93             raise HdfsAtomicWriteError('Atomic write to {} failed'.format(self.path))
94 
95 
96 class PlainFormat(luigi.format.Format):
97 
98     input = 'bytes'
99     output = 'hdfs'
100 
101     def hdfs_writer(self, path):
102         return self.pipe_writer(path)
103 
104     def hdfs_reader(self, path):
105         return self.pipe_reader(path)
106 
107     def pipe_reader(self, path):
108         return HdfsReadPipe(path)
109 
110     def pipe_writer(self, output_pipe):
111         return HdfsAtomicWritePipe(output_pipe)
112 
113 
114 class PlainDirFormat(luigi.format.Format):
115 
116     input = 'bytes'
117     output = 'hdfs'
118 
119     def hdfs_writer(self, path):
120         return self.pipe_writer(path)
121 
122     def hdfs_reader(self, path):
123         return self.pipe_reader(path)
124 
125     def pipe_reader(self, path):
126         # exclude underscore-prefixedfiles/folders (created by MapReduce)
127         return HdfsReadPipe("%s/[^_]*" % path)
128 
129     def pipe_writer(self, path):
130         return HdfsAtomicWriteDirPipe(path)
131 
132 
133 Plain = PlainFormat()
134 PlainDir = PlainDirFormat()
135 
136 
137 class CompatibleHdfsFormat(luigi.format.Format):
138 
139     output = 'hdfs'
140 
141     def __init__(self, writer, reader, input=None):
142         if input is not None:
143             self.input = input
144 
145         self.reader = reader
146         self.writer = writer
147 
148     def pipe_writer(self, output):
149         return self.writer(output)
150 
151     def pipe_reader(self, input):
152         return self.reader(input)
153 
154     def hdfs_writer(self, output):
155         return self.writer(output)
156 
157     def hdfs_reader(self, input):
158         return self.reader(input)
159 
160     # __getstate__/__setstate__ needed for pickling, because self.reader and
161     # self.writer may be unpickleable instance methods of another format class.
162     # This was mainly to support pickling of standard HdfsTarget instances.
163 
164     def __getstate__(self):
165         d = self.__dict__.copy()
166         for attr in ('reader', 'writer'):
167             method = getattr(self, attr)
168             try:
169                 # if instance method, pickle instance and method name
170                 d[attr] = method.__self__, method.__func__.__name__
171             except AttributeError:
172                 pass  # not an instance method
173         return d
174 
175     def __setstate__(self, d):
176         self.__dict__ = d
177         for attr in ('reader', 'writer'):
178             try:
179                 method_self, method_name = d[attr]
180             except ValueError:
181                 continue
182             method = getattr(method_self, method_name)
183             setattr(self, attr, method)
184
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/luigi/contrib/hdfs/format.py b/luigi/contrib/hdfs/format.py
--- a/luigi/contrib/hdfs/format.py
+++ b/luigi/contrib/hdfs/format.py
@@ -49,9 +49,13 @@
     def close(self):
         super(HdfsAtomicWritePipe, self).close()
         try:
-            remove(self.path)
-        except HDFSCliError:
-            pass
+            if exists(self.path):
+                remove(self.path)
+        except Exception as ex:
+            if isinstance(ex, HDFSCliError) or ex.args[0].contains("FileNotFoundException"):
+                pass
+            else:
+                raise ex
         if not all(result['result'] for result in rename(self.tmppath, self.path) or []):
             raise HdfsAtomicWriteError('Atomic write to {} failed'.format(self.path))

{"golden_diff": "diff --git a/luigi/contrib/hdfs/format.py b/luigi/contrib/hdfs/format.py\n--- a/luigi/contrib/hdfs/format.py\n+++ b/luigi/contrib/hdfs/format.py\n@@ -49,9 +49,13 @@\n     def close(self):\n         super(HdfsAtomicWritePipe, self).close()\n         try:\n-            remove(self.path)\n-        except HDFSCliError:\n-            pass\n+            if exists(self.path):\n+                remove(self.path)\n+        except Exception as ex:\n+            if isinstance(ex, HDFSCliError) or ex.args[0].contains(\"FileNotFoundException\"):\n+                pass\n+            else:\n+                raise ex\n         if not all(result['result'] for result in rename(self.tmppath, self.path) or []):\n             raise HdfsAtomicWriteError('Atomic write to {} failed'.format(self.path))\n", "issue": "HdfsTarget is not workging in luigi > 2.6.1\nThis code runs correct under luigi <= 2.6.1:\r\n\r\n```python\r\nfrom luigi.contrib import hdfs\r\nimport luigi.format\r\n\r\n# just to be sure that realy the right version is used...\r\nimport pkg_resources\r\nprint \"luigi ==\", pkg_resources.get_distribution(\"luigi\").version\r\nprint \"snakebite ==\", pkg_resources.get_distribution(\"snakebite\").version\r\n\r\ndestination_file = '/tmp/test/file.gz'\r\n\r\ntarget = hdfs.HdfsTarget(path=destination_file, format=luigi.format.Gzip)\r\n\r\nif target.exists():\r\n    target.remove(skip_trash=False)\r\n\r\nfsobj = target.open('w')\r\nfsobj.write('lol3\\n')\r\nfsobj.close()\r\n```\r\n\r\nwith luigi 2.6.2 or 2.7.0 it breaks:\r\n```python\r\nluigi == 2.7.0\r\nsnakebite == 2.11.0\r\nTraceback (most recent call last):\r\n  File \"/opt/tests/hdfs_target.py\", line 18, in <module>\r\n    fsobj.close()\r\n  File \"/opt/python-2.7.10/lib/python2.7/site-packages/luigi/format.py\", line 224, in close\r\n    self._output_pipe.close()\r\n  File \"/opt/python-2.7.10/lib/python2.7/site-packages/luigi/contrib/hdfs/format.py\", line 51, in close\r\n    remove(self.path)\r\n  File \"/opt/python-2.7.10/lib/python2.7/site-packages/luigi/contrib/hdfs/clients.py\", line 62, in result\r\n    return getattr(get_autoconfig_client(), method_name)(*args, **kwargs)\r\n  File \"/opt/python-2.7.10/lib/python2.7/site-packages/luigi/contrib/hdfs/snakebite_client.py\", line 140, in remove\r\n    return list(self.get_bite().delete(self.list_path(path), recurse=recursive))\r\n  File \"/opt/python-2.7.10/lib/python2.7/site-packages/snakebite/client.py\", line 1540, in wrapped\r\n    yield results.next()\r\n  File \"/opt/python-2.7.10/lib/python2.7/site-packages/snakebite/client.py\", line 508, in delete\r\n    for item in self._find_items(paths, processor, include_toplevel=True):\r\n  File \"/opt/python-2.7.10/lib/python2.7/site-packages/snakebite/client.py\", line 1216, in _find_items\r\n    raise FileNotFoundException(\"`%s': No such file or directory\" % path)\r\nsnakebite.errors.FileNotFoundException: `/tmp/test/file.gz': No such file or directory\r\n```\r\n\r\nclient.cfg:\r\n```\r\n[hdfs]\r\nclient = snakebite\r\nsnakebite_autoconfig = True\r\ntmp_dir: /x/tmp\r\n```\r\n\r\nThe file is correct in form and content written in \"tmp_dir\" (/x/tmp/username/test/file.gz-luigitemp-951771388).\n", "before_files": [{"content": "import logging\nimport os\n\nimport luigi.format\nfrom luigi.contrib.hdfs.config import load_hadoop_cmd\nfrom luigi.contrib.hdfs import config as hdfs_config\nfrom luigi.contrib.hdfs.clients import remove, rename, mkdir, listdir, exists\nfrom luigi.contrib.hdfs.error import HDFSCliError\n\nlogger = logging.getLogger('luigi-interface')\n\n\nclass HdfsAtomicWriteError(IOError):\n    pass\n\n\nclass HdfsReadPipe(luigi.format.InputPipeProcessWrapper):\n\n    def __init__(self, path):\n        super(HdfsReadPipe, self).__init__(load_hadoop_cmd() + ['fs', '-cat', path])\n\n\nclass HdfsAtomicWritePipe(luigi.format.OutputPipeProcessWrapper):\n    \"\"\"\n    File like object for writing to HDFS\n\n    The referenced file is first written to a temporary location and then\n    renamed to final location on close(). If close() isn't called\n    the temporary file will be cleaned up when this object is\n    garbage collected\n\n    TODO: if this is buggy, change it so it first writes to a\n    local temporary file and then uploads it on completion\n    \"\"\"\n\n    def __init__(self, path):\n        self.path = path\n        self.tmppath = hdfs_config.tmppath(self.path)\n        parent_dir = os.path.dirname(self.tmppath)\n        mkdir(parent_dir, parents=True, raise_if_exists=False)\n        super(HdfsAtomicWritePipe, self).__init__(load_hadoop_cmd() + ['fs', '-put', '-', self.tmppath])\n\n    def abort(self):\n        logger.info(\"Aborting %s('%s'). Removing temporary file '%s'\",\n                    self.__class__.__name__, self.path, self.tmppath)\n        super(HdfsAtomicWritePipe, self).abort()\n        remove(self.tmppath, skip_trash=True)\n\n    def close(self):\n        super(HdfsAtomicWritePipe, self).close()\n        try:\n            remove(self.path)\n        except HDFSCliError:\n            pass\n        if not all(result['result'] for result in rename(self.tmppath, self.path) or []):\n            raise HdfsAtomicWriteError('Atomic write to {} failed'.format(self.path))\n\n\nclass HdfsAtomicWriteDirPipe(luigi.format.OutputPipeProcessWrapper):\n    \"\"\"\n    Writes a data<data_extension> file to a directory at <path>.\n    \"\"\"\n\n    def __init__(self, path, data_extension=\"\"):\n        self.path = path\n        self.tmppath = hdfs_config.tmppath(self.path)\n        self.datapath = self.tmppath + (\"/data%s\" % data_extension)\n        super(HdfsAtomicWriteDirPipe, self).__init__(load_hadoop_cmd() + ['fs', '-put', '-', self.datapath])\n\n    def abort(self):\n        logger.info(\"Aborting %s('%s'). Removing temporary dir '%s'\",\n                    self.__class__.__name__, self.path, self.tmppath)\n        super(HdfsAtomicWriteDirPipe, self).abort()\n        remove(self.tmppath, skip_trash=True)\n\n    def close(self):\n        super(HdfsAtomicWriteDirPipe, self).close()\n        try:\n            if exists(self.path):\n                remove(self.path)\n        except Exception as ex:\n            if isinstance(ex, HDFSCliError) or ex.args[0].contains(\"FileNotFoundException\"):\n                pass\n            else:\n                raise ex\n\n        # it's unlikely to fail in this way but better safe than sorry\n        if not all(result['result'] for result in rename(self.tmppath, self.path) or []):\n            raise HdfsAtomicWriteError('Atomic write to {} failed'.format(self.path))\n\n        if os.path.basename(self.tmppath) in map(os.path.basename, listdir(self.path)):\n            remove(self.path)\n            raise HdfsAtomicWriteError('Atomic write to {} failed'.format(self.path))\n\n\nclass PlainFormat(luigi.format.Format):\n\n    input = 'bytes'\n    output = 'hdfs'\n\n    def hdfs_writer(self, path):\n        return self.pipe_writer(path)\n\n    def hdfs_reader(self, path):\n        return self.pipe_reader(path)\n\n    def pipe_reader(self, path):\n        return HdfsReadPipe(path)\n\n    def pipe_writer(self, output_pipe):\n        return HdfsAtomicWritePipe(output_pipe)\n\n\nclass PlainDirFormat(luigi.format.Format):\n\n    input = 'bytes'\n    output = 'hdfs'\n\n    def hdfs_writer(self, path):\n        return self.pipe_writer(path)\n\n    def hdfs_reader(self, path):\n        return self.pipe_reader(path)\n\n    def pipe_reader(self, path):\n        # exclude underscore-prefixedfiles/folders (created by MapReduce)\n        return HdfsReadPipe(\"%s/[^_]*\" % path)\n\n    def pipe_writer(self, path):\n        return HdfsAtomicWriteDirPipe(path)\n\n\nPlain = PlainFormat()\nPlainDir = PlainDirFormat()\n\n\nclass CompatibleHdfsFormat(luigi.format.Format):\n\n    output = 'hdfs'\n\n    def __init__(self, writer, reader, input=None):\n        if input is not None:\n            self.input = input\n\n        self.reader = reader\n        self.writer = writer\n\n    def pipe_writer(self, output):\n        return self.writer(output)\n\n    def pipe_reader(self, input):\n        return self.reader(input)\n\n    def hdfs_writer(self, output):\n        return self.writer(output)\n\n    def hdfs_reader(self, input):\n        return self.reader(input)\n\n    # __getstate__/__setstate__ needed for pickling, because self.reader and\n    # self.writer may be unpickleable instance methods of another format class.\n    # This was mainly to support pickling of standard HdfsTarget instances.\n\n    def __getstate__(self):\n        d = self.__dict__.copy()\n        for attr in ('reader', 'writer'):\n            method = getattr(self, attr)\n            try:\n                # if instance method, pickle instance and method name\n                d[attr] = method.__self__, method.__func__.__name__\n            except AttributeError:\n                pass  # not an instance method\n        return d\n\n    def __setstate__(self, d):\n        self.__dict__ = d\n        for attr in ('reader', 'writer'):\n            try:\n                method_self, method_name = d[attr]\n            except ValueError:\n                continue\n            method = getattr(method_self, method_name)\n            setattr(self, attr, method)\n", "path": "luigi/contrib/hdfs/format.py"}], "after_files": [{"content": "import logging\nimport os\n\nimport luigi.format\nfrom luigi.contrib.hdfs.config import load_hadoop_cmd\nfrom luigi.contrib.hdfs import config as hdfs_config\nfrom luigi.contrib.hdfs.clients import remove, rename, mkdir, listdir, exists\nfrom luigi.contrib.hdfs.error import HDFSCliError\n\nlogger = logging.getLogger('luigi-interface')\n\n\nclass HdfsAtomicWriteError(IOError):\n    pass\n\n\nclass HdfsReadPipe(luigi.format.InputPipeProcessWrapper):\n\n    def __init__(self, path):\n        super(HdfsReadPipe, self).__init__(load_hadoop_cmd() + ['fs', '-cat', path])\n\n\nclass HdfsAtomicWritePipe(luigi.format.OutputPipeProcessWrapper):\n    \"\"\"\n    File like object for writing to HDFS\n\n    The referenced file is first written to a temporary location and then\n    renamed to final location on close(). If close() isn't called\n    the temporary file will be cleaned up when this object is\n    garbage collected\n\n    TODO: if this is buggy, change it so it first writes to a\n    local temporary file and then uploads it on completion\n    \"\"\"\n\n    def __init__(self, path):\n        self.path = path\n        self.tmppath = hdfs_config.tmppath(self.path)\n        parent_dir = os.path.dirname(self.tmppath)\n        mkdir(parent_dir, parents=True, raise_if_exists=False)\n        super(HdfsAtomicWritePipe, self).__init__(load_hadoop_cmd() + ['fs', '-put', '-', self.tmppath])\n\n    def abort(self):\n        logger.info(\"Aborting %s('%s'). Removing temporary file '%s'\",\n                    self.__class__.__name__, self.path, self.tmppath)\n        super(HdfsAtomicWritePipe, self).abort()\n        remove(self.tmppath, skip_trash=True)\n\n    def close(self):\n        super(HdfsAtomicWritePipe, self).close()\n        try:\n            if exists(self.path):\n                remove(self.path)\n        except Exception as ex:\n            if isinstance(ex, HDFSCliError) or ex.args[0].contains(\"FileNotFoundException\"):\n                pass\n            else:\n                raise ex\n        if not all(result['result'] for result in rename(self.tmppath, self.path) or []):\n            raise HdfsAtomicWriteError('Atomic write to {} failed'.format(self.path))\n\n\nclass HdfsAtomicWriteDirPipe(luigi.format.OutputPipeProcessWrapper):\n    \"\"\"\n    Writes a data<data_extension> file to a directory at <path>.\n    \"\"\"\n\n    def __init__(self, path, data_extension=\"\"):\n        self.path = path\n        self.tmppath = hdfs_config.tmppath(self.path)\n        self.datapath = self.tmppath + (\"/data%s\" % data_extension)\n        super(HdfsAtomicWriteDirPipe, self).__init__(load_hadoop_cmd() + ['fs', '-put', '-', self.datapath])\n\n    def abort(self):\n        logger.info(\"Aborting %s('%s'). Removing temporary dir '%s'\",\n                    self.__class__.__name__, self.path, self.tmppath)\n        super(HdfsAtomicWriteDirPipe, self).abort()\n        remove(self.tmppath, skip_trash=True)\n\n    def close(self):\n        super(HdfsAtomicWriteDirPipe, self).close()\n        try:\n            if exists(self.path):\n                remove(self.path)\n        except Exception as ex:\n            if isinstance(ex, HDFSCliError) or ex.args[0].contains(\"FileNotFoundException\"):\n                pass\n            else:\n                raise ex\n\n        # it's unlikely to fail in this way but better safe than sorry\n        if not all(result['result'] for result in rename(self.tmppath, self.path) or []):\n            raise HdfsAtomicWriteError('Atomic write to {} failed'.format(self.path))\n\n        if os.path.basename(self.tmppath) in map(os.path.basename, listdir(self.path)):\n            remove(self.path)\n            raise HdfsAtomicWriteError('Atomic write to {} failed'.format(self.path))\n\n\nclass PlainFormat(luigi.format.Format):\n\n    input = 'bytes'\n    output = 'hdfs'\n\n    def hdfs_writer(self, path):\n        return self.pipe_writer(path)\n\n    def hdfs_reader(self, path):\n        return self.pipe_reader(path)\n\n    def pipe_reader(self, path):\n        return HdfsReadPipe(path)\n\n    def pipe_writer(self, output_pipe):\n        return HdfsAtomicWritePipe(output_pipe)\n\n\nclass PlainDirFormat(luigi.format.Format):\n\n    input = 'bytes'\n    output = 'hdfs'\n\n    def hdfs_writer(self, path):\n        return self.pipe_writer(path)\n\n    def hdfs_reader(self, path):\n        return self.pipe_reader(path)\n\n    def pipe_reader(self, path):\n        # exclude underscore-prefixedfiles/folders (created by MapReduce)\n        return HdfsReadPipe(\"%s/[^_]*\" % path)\n\n    def pipe_writer(self, path):\n        return HdfsAtomicWriteDirPipe(path)\n\n\nPlain = PlainFormat()\nPlainDir = PlainDirFormat()\n\n\nclass CompatibleHdfsFormat(luigi.format.Format):\n\n    output = 'hdfs'\n\n    def __init__(self, writer, reader, input=None):\n        if input is not None:\n            self.input = input\n\n        self.reader = reader\n        self.writer = writer\n\n    def pipe_writer(self, output):\n        return self.writer(output)\n\n    def pipe_reader(self, input):\n        return self.reader(input)\n\n    def hdfs_writer(self, output):\n        return self.writer(output)\n\n    def hdfs_reader(self, input):\n        return self.reader(input)\n\n    # __getstate__/__setstate__ needed for pickling, because self.reader and\n    # self.writer may be unpickleable instance methods of another format class.\n    # This was mainly to support pickling of standard HdfsTarget instances.\n\n    def __getstate__(self):\n        d = self.__dict__.copy()\n        for attr in ('reader', 'writer'):\n            method = getattr(self, attr)\n            try:\n                # if instance method, pickle instance and method name\n                d[attr] = method.__self__, method.__func__.__name__\n            except AttributeError:\n                pass  # not an instance method\n        return d\n\n    def __setstate__(self, d):\n        self.__dict__ = d\n        for attr in ('reader', 'writer'):\n            try:\n                method_self, method_name = d[attr]\n            except ValueError:\n                continue\n            method = getattr(method_self, method_name)\n            setattr(self, attr, method)\n", "path": "luigi/contrib/hdfs/format.py"}]}