Author: tgraves Date: Fri Aug 17 19:47:35 2012 New Revision: 1374407 URL: http://svn.apache.org/viewvc?rev=1374407&view=rev Log: MAPREDUCE-4549. Distributed cache conflicts breaks backwards compatability (Robert Evans via tgraves)
Modified: hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRApps.java hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapreduce/v2/util/TestMRApps.java Modified: hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt?rev=1374407&r1=1374406&r2=1374407&view=diff ============================================================================== --- hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt (original) +++ hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt Fri Aug 17 19:47:35 2012 @@ -392,6 +392,9 @@ Release 0.23.3 - UNRELEASED for compatibility reasons is creating incorrect counter name. (Jarek Jarcec Cecho via tomwhite) + MAPREDUCE-4549. Distributed cache conflicts breaks backwards compatability + (Robert Evans via tgraves) + Release 0.23.2 - UNRELEASED INCOMPATIBLE CHANGES Modified: hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRApps.java URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRApps.java?rev=1374407&r1=1374406&r2=1374407&view=diff ============================================================================== --- hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRApps.java (original) +++ hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRApps.java Fri Aug 17 19:47:35 2012 @@ -30,6 +30,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.hadoop.classification.InterfaceAudience.Private; import org.apache.hadoop.classification.InterfaceStability.Unstable; import org.apache.hadoop.conf.Configuration; @@ -48,14 +50,15 @@ import org.apache.hadoop.mapreduce.v2.ap import org.apache.hadoop.mapreduce.v2.api.records.TaskType; import org.apache.hadoop.yarn.ContainerLogAppender; import org.apache.hadoop.yarn.YarnException; -import org.apache.hadoop.yarn.api.ApplicationConstants.Environment; import org.apache.hadoop.yarn.api.ApplicationConstants; +import org.apache.hadoop.yarn.api.ApplicationConstants.Environment; import org.apache.hadoop.yarn.api.records.LocalResource; import org.apache.hadoop.yarn.api.records.LocalResourceType; import org.apache.hadoop.yarn.api.records.LocalResourceVisibility; import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.util.Apps; import org.apache.hadoop.yarn.util.BuilderUtils; +import org.apache.hadoop.yarn.util.ConverterUtils; /** * Helper class for MR applications @@ -63,6 +66,8 @@ import org.apache.hadoop.yarn.util.Build @Private @Unstable public class MRApps extends Apps { + private static final Log LOG = LogFactory.getLog(MRApps.class); + public static String toString(JobId jid) { return jid.toString(); } @@ -263,6 +268,23 @@ public class MRApps extends Apps { DistributedCache.getFileClassPaths(conf)); } + private static String getResourceDescription(LocalResourceType type) { + if(type == LocalResourceType.ARCHIVE) { + return "cache archive (" + MRJobConfig.CACHE_ARCHIVES + ") "; + } + return "cache file (" + MRJobConfig.CACHE_FILES + ") "; + } + + private static String toString(org.apache.hadoop.yarn.api.records.URL url) { + StringBuffer b = new StringBuffer(); + b.append(url.getScheme()).append("://").append(url.getHost()); + if(url.getPort() >= 0) { + b.append(":").append(url.getPort()); + } + b.append(url.getFile()); + return b.toString(); + } + // TODO - Move this to MR! // Use TaskDistributedCacheManager.CacheFiles.makeCacheFiles(URI[], // long[], boolean[], Path[], FileType) @@ -308,6 +330,17 @@ public class MRApps extends Apps { throw new IllegalArgumentException("Resource name must be relative"); } String linkName = name.toUri().getPath(); + LocalResource orig = localResources.get(linkName); + org.apache.hadoop.yarn.api.records.URL url = + ConverterUtils.getYarnUrlFromURI(p.toUri()); + if(orig != null && !orig.getResource().equals(url)) { + LOG.warn( + getResourceDescription(orig.getType()) + + toString(orig.getResource()) + " conflicts with " + + getResourceDescription(type) + toString(url) + + " This will be an error in Hadoop 2.0"); + continue; + } localResources.put( linkName, BuilderUtils.newLocalResource( Modified: hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapreduce/v2/util/TestMRApps.java URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapreduce/v2/util/TestMRApps.java?rev=1374407&r1=1374406&r2=1374407&view=diff ============================================================================== --- hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapreduce/v2/util/TestMRApps.java (original) +++ hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapreduce/v2/util/TestMRApps.java Fri Aug 17 19:47:35 2012 @@ -19,23 +19,30 @@ package org.apache.hadoop.mapreduce.v2.util; import java.io.IOException; +import java.net.URI; import java.util.HashMap; import java.util.Map; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.FilterFileSystem; +import org.apache.hadoop.fs.Path; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.JobID; import org.apache.hadoop.mapreduce.MRJobConfig; +import org.apache.hadoop.mapreduce.filecache.DistributedCache; import org.apache.hadoop.mapreduce.v2.api.records.JobId; import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId; import org.apache.hadoop.mapreduce.v2.api.records.TaskId; import org.apache.hadoop.mapreduce.v2.api.records.TaskType; -import org.apache.hadoop.mapreduce.v2.util.MRApps; import org.apache.hadoop.yarn.api.records.ApplicationId; +import org.apache.hadoop.yarn.api.records.LocalResource; +import org.apache.hadoop.yarn.api.records.LocalResourceType; import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider; - import org.junit.Test; + import static org.junit.Assert.*; +import static org.mockito.Mockito.*; public class TestMRApps { @@ -168,5 +175,138 @@ public class TestMRApps { assertNotSame("MAPREDUCE_JOB_USER_CLASSPATH_FIRST false, but taking effect!", env_str.indexOf("$PWD:job.jar"), 0); } - + + @Test + public void testSetupDistributedCacheEmpty() throws IOException { + Configuration conf = new Configuration(); + Map<String, LocalResource> localResources = new HashMap<String, LocalResource>(); + MRApps.setupDistributedCache(conf, localResources); + assertTrue("Empty Config did not produce an empty list of resources", + localResources.isEmpty()); + } + + @SuppressWarnings("deprecation") + @Test + public void testSetupDistributedCacheConflicts() throws Exception { + Configuration conf = new Configuration(); + conf.setClass("fs.mockfs.impl", MockFileSystem.class, FileSystem.class); + + URI mockUri = URI.create("mockfs://mock/"); + FileSystem mockFs = ((FilterFileSystem)FileSystem.get(mockUri, conf)) + .getRawFileSystem(); + + URI archive = new URI("mockfs://mock/tmp/something.zip#something"); + Path archivePath = new Path(archive); + URI file = new URI("mockfs://mock/tmp/something.txt#something"); + Path filePath = new Path(file); + + when(mockFs.resolvePath(archivePath)).thenReturn(archivePath); + when(mockFs.resolvePath(filePath)).thenReturn(filePath); + + DistributedCache.addCacheArchive(archive, conf); + conf.set(MRJobConfig.CACHE_ARCHIVES_TIMESTAMPS, "10"); + conf.set(MRJobConfig.CACHE_ARCHIVES_SIZES, "10"); + conf.set(MRJobConfig.CACHE_ARCHIVES_VISIBILITIES, "true"); + DistributedCache.addCacheFile(file, conf); + conf.set(MRJobConfig.CACHE_FILE_TIMESTAMPS, "11"); + conf.set(MRJobConfig.CACHE_FILES_SIZES, "11"); + conf.set(MRJobConfig.CACHE_FILE_VISIBILITIES, "true"); + Map<String, LocalResource> localResources = + new HashMap<String, LocalResource>(); + MRApps.setupDistributedCache(conf, localResources); + + assertEquals(1, localResources.size()); + LocalResource lr = localResources.get("something"); + //Archive wins + assertNotNull(lr); + assertEquals(10l, lr.getSize()); + assertEquals(10l, lr.getTimestamp()); + assertEquals(LocalResourceType.ARCHIVE, lr.getType()); + } + + @SuppressWarnings("deprecation") + @Test + public void testSetupDistributedCacheConflictsFiles() throws Exception { + Configuration conf = new Configuration(); + conf.setClass("fs.mockfs.impl", MockFileSystem.class, FileSystem.class); + + URI mockUri = URI.create("mockfs://mock/"); + FileSystem mockFs = ((FilterFileSystem)FileSystem.get(mockUri, conf)) + .getRawFileSystem(); + + URI file = new URI("mockfs://mock/tmp/something.zip#something"); + Path filePath = new Path(file); + URI file2 = new URI("mockfs://mock/tmp/something.txt#something"); + Path file2Path = new Path(file); + + when(mockFs.resolvePath(filePath)).thenReturn(filePath); + when(mockFs.resolvePath(file2Path)).thenReturn(file2Path); + + DistributedCache.addCacheFile(file, conf); + DistributedCache.addCacheFile(file2, conf); + conf.set(MRJobConfig.CACHE_FILE_TIMESTAMPS, "10,11"); + conf.set(MRJobConfig.CACHE_FILES_SIZES, "10,11"); + conf.set(MRJobConfig.CACHE_FILE_VISIBILITIES, "true,true"); + Map<String, LocalResource> localResources = + new HashMap<String, LocalResource>(); + MRApps.setupDistributedCache(conf, localResources); + + assertEquals(1, localResources.size()); + LocalResource lr = localResources.get("something"); + //First one wins + assertNotNull(lr); + assertEquals(10l, lr.getSize()); + assertEquals(10l, lr.getTimestamp()); + assertEquals(LocalResourceType.FILE, lr.getType()); + } + + @SuppressWarnings("deprecation") + @Test + public void testSetupDistributedCache() throws Exception { + Configuration conf = new Configuration(); + conf.setClass("fs.mockfs.impl", MockFileSystem.class, FileSystem.class); + + URI mockUri = URI.create("mockfs://mock/"); + FileSystem mockFs = ((FilterFileSystem)FileSystem.get(mockUri, conf)) + .getRawFileSystem(); + + URI archive = new URI("mockfs://mock/tmp/something.zip"); + Path archivePath = new Path(archive); + URI file = new URI("mockfs://mock/tmp/something.txt#something"); + Path filePath = new Path(file); + + when(mockFs.resolvePath(archivePath)).thenReturn(archivePath); + when(mockFs.resolvePath(filePath)).thenReturn(filePath); + + DistributedCache.addCacheArchive(archive, conf); + conf.set(MRJobConfig.CACHE_ARCHIVES_TIMESTAMPS, "10"); + conf.set(MRJobConfig.CACHE_ARCHIVES_SIZES, "10"); + conf.set(MRJobConfig.CACHE_ARCHIVES_VISIBILITIES, "true"); + DistributedCache.addCacheFile(file, conf); + conf.set(MRJobConfig.CACHE_FILE_TIMESTAMPS, "11"); + conf.set(MRJobConfig.CACHE_FILES_SIZES, "11"); + conf.set(MRJobConfig.CACHE_FILE_VISIBILITIES, "true"); + Map<String, LocalResource> localResources = + new HashMap<String, LocalResource>(); + MRApps.setupDistributedCache(conf, localResources); + assertEquals(2, localResources.size()); + LocalResource lr = localResources.get("something.zip"); + assertNotNull(lr); + assertEquals(10l, lr.getSize()); + assertEquals(10l, lr.getTimestamp()); + assertEquals(LocalResourceType.ARCHIVE, lr.getType()); + lr = localResources.get("something"); + assertNotNull(lr); + assertEquals(11l, lr.getSize()); + assertEquals(11l, lr.getTimestamp()); + assertEquals(LocalResourceType.FILE, lr.getType()); + } + + static class MockFileSystem extends FilterFileSystem { + MockFileSystem() { + super(mock(FileSystem.class)); + } + public void initialize(URI name, Configuration conf) throws IOException {} + } + }