We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
HdfsAtomicWritePipe silently fails when target file exists
Lately due to some pipeline issues I've ended up re-running a lot of jobs while their output exists. They seem to run fine but the output doesn't change or even get an updated modified time. I tracked this down to the atomic write pipe just silently failing when the move does nothing. I use the snakebite client with hdfs fallback. I believe this is happening in the snakebite client.

Demonstration code:

```
import luigi.contrib.hdfs

target = luigi.contrib.hdfs.HdfsTarget('/tmp/test.txt')
with target.open('w') as fobj:
    fobj.write('test1')

try:
    with target.open('w') as fobj:
        fobj.write('test2')
finally:
    with target.open() as fobj:
        print '\ncontents:', fobj.read()
    target.remove()
```
I would expect to either see `test2` printed. At the very least, I'd expect to see an error message if it prints `test1`, as this means the second write didn't work. Instead I see `test1` and no error message. So it looks like I successfully wrote `test2` to hdfs when I didn't.

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

{"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@@ -3,11 +3,16 @@\n import os\n from luigi.contrib.hdfs.config import load_hadoop_cmd\n from luigi.contrib.hdfs import config as hdfs_config\n-from luigi.contrib.hdfs.clients import remove, rename, mkdir\n+from luigi.contrib.hdfs.clients import remove, rename, mkdir, listdir\n+from luigi.contrib.hdfs.error import HDFSCliError\n \n logger = logging.getLogger('luigi-interface')\n \n \n+class HdfsAtomicWriteError(IOError):\n+    pass\n+\n+\n class HdfsReadPipe(luigi.format.InputPipeProcessWrapper):\n \n     def __init__(self, path):\n@@ -42,7 +47,12 @@\n \n     def close(self):\n         super(HdfsAtomicWritePipe, self).close()\n-        rename(self.tmppath, self.path)\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 \n class HdfsAtomicWriteDirPipe(luigi.format.OutputPipeProcessWrapper):\n@@ -64,7 +74,18 @@\n \n     def close(self):\n         super(HdfsAtomicWriteDirPipe, self).close()\n-        rename(self.tmppath, self.path)\n+        try:\n+            remove(self.path)\n+        except HDFSCliError:\n+            pass\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 \n class PlainFormat(luigi.format.Format):\n", "issue": "HdfsAtomicWritePipe silently fails when target file exists\nLately due to some pipeline issues I've ended up re-running a lot of jobs while their output exists. They seem to run fine but the output doesn't change or even get an updated modified time. I tracked this down to the atomic write pipe just silently failing when the move does nothing. I use the snakebite client with hdfs fallback. I believe this is happening in the snakebite client.\r\n\r\nDemonstration code:\r\n\r\n```\r\nimport luigi.contrib.hdfs\r\n\r\ntarget = luigi.contrib.hdfs.HdfsTarget('/tmp/test.txt')\r\nwith target.open('w') as fobj:\r\n    fobj.write('test1')\r\n\r\ntry:\r\n    with target.open('w') as fobj:\r\n        fobj.write('test2')\r\nfinally:\r\n    with target.open() as fobj:\r\n        print '\\ncontents:', fobj.read()\r\n    target.remove()\r\n```\r\nI would expect to either see `test2` printed. At the very least, I'd expect to see an error message if it prints `test1`, as this means the second write didn't work. Instead I see `test1` and no error message. So it looks like I successfully wrote `test2` to hdfs when I didn't.\n", "before_files": [{"content": "import luigi.format\nimport logging\nimport os\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\n\nlogger = logging.getLogger('luigi-interface')\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        rename(self.tmppath, 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        rename(self.tmppath, 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 luigi.format\nimport logging\nimport os\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\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            remove(self.path)\n        except HDFSCliError:\n            pass\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"}]}