Repository: ambari Updated Branches: refs/heads/trunk 730c081d7 -> 75e982a9d
AMBARI-9990. CopyFromLocal failed to copy Tez tarball to HDFS failed because multiple processes tried to copy to the same destination simultaneously (alejandro) Project: http://git-wip-us.apache.org/repos/asf/ambari/repo Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/75e982a9 Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/75e982a9 Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/75e982a9 Branch: refs/heads/trunk Commit: 75e982a9d84774929fb225bb0bdebedf043f9c08 Parents: 730c081 Author: Alejandro Fernandez <[email protected]> Authored: Mon Mar 9 15:29:15 2015 -0700 Committer: Alejandro Fernandez <[email protected]> Committed: Tue Mar 10 18:10:21 2015 -0700 ---------------------------------------------------------------------- .../functions/dynamic_variable_interpretation.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ambari/blob/75e982a9/ambari-common/src/main/python/resource_management/libraries/functions/dynamic_variable_interpretation.py ---------------------------------------------------------------------- diff --git a/ambari-common/src/main/python/resource_management/libraries/functions/dynamic_variable_interpretation.py b/ambari-common/src/main/python/resource_management/libraries/functions/dynamic_variable_interpretation.py index 00b8d70..1e70219 100644 --- a/ambari-common/src/main/python/resource_management/libraries/functions/dynamic_variable_interpretation.py +++ b/ambari-common/src/main/python/resource_management/libraries/functions/dynamic_variable_interpretation.py @@ -23,6 +23,7 @@ import os import glob import re import tempfile +import uuid from resource_management.libraries.functions.default import default from resource_management.libraries.functions.format import format from resource_management.libraries.resources.copy_from_local import CopyFromLocal @@ -101,17 +102,35 @@ def _copy_files(source_and_dest_pairs, component_user, file_owner, group_owner, mode=0555 ) + # Because CopyFromLocal does not guarantee synchronization, it's possible for two processes to first attempt to + # copy the file to a temporary location, then process 2 fails because the temporary file was already created by + # process 1, so process 2 tries to clean up by deleting the temporary file, and then process 1 + # cannot finish the copy to the final destination, and both fail! + # For this reason, the file name on the destination must be unique, and we then rename it to the intended value. + # The rename operation is synchronized by the Namenode. + orig_dest_file_name = os.path.split(destination)[1] + unique_string = str(uuid.uuid4())[:8] + new_dest_file_name = orig_dest_file_name + "." + unique_string + new_destination = os.path.join(destination_dir, new_dest_file_name) CopyFromLocal(source, mode=0444, owner=file_owner, group=group_owner, user=params.hdfs_user, # this will be the user to run the commands as dest_dir=destination_dir, + dest_file=new_dest_file_name, kinnit_if_needed=kinit_if_needed, hdfs_user=params.hdfs_user, hadoop_bin_dir=params.hadoop_bin_dir, hadoop_conf_dir=params.hadoop_conf_dir ) + + mv_command = format("fs -mv {new_destination} {destination}") + ExecuteHadoop(mv_command, + user=params.hdfs_user, + bin_dir=params.hadoop_bin_dir, + conf_dir=params.hadoop_conf_dir + ) except Exception, e: Logger.error("Failed to copy file. Source: %s, Destination: %s. Error: %s" % (source, destination, e.message)) return_value = 1
