ContainerLocalizer { public static void main(String[] argv) throws Throwable { Thread.setDefaultUncaughtExceptionHandler(new YarnUncaughtExceptionHandler()); int nRet = 0; try { String user = argv[0]; String appId = argv[1]; String locId = argv[2]; InetSocketAddress nmAddr = new InetSocketAddress(argv[3], Integer.parseInt(argv[4])); String[] sLocaldirs = Arrays.copyOfRange(argv, 5, argv.length); ArrayList<Path> localDirs = new ArrayList<Path>(sLocaldirs.length); for (String sLocaldir : sLocaldirs) { localDirs.add(new Path(sLocaldir)); } final String uid = UserGroupInformation.getCurrentUser().getShortUserName(); if (!user.equals(uid)) { LOG.warn("Localization running as " + uid + " not " + user); } ContainerLocalizer localizer = new ContainerLocalizer(FileContext.getLocalFSFileContext(), user, appId, locId, localDirs, RecordFactoryProvider.getRecordFactory(null)); localizer.runLocalization(nmAddr); return; } catch (Throwable e) { e.printStackTrace(System.out); LOG.error("Exception in main:", e); nRet = -1; } finally { System.exit(nRet); } } ContainerLocalizer(FileContext lfs, String user, String appId,
      String localizerId, List<Path> localDirs,
      RecordFactory recordFactory); @Private @VisibleForTesting LocalizationProtocol getProxy(final InetSocketAddress nmAddr); @SuppressWarnings("deprecation") void runLocalization(final InetSocketAddress nmAddr); static List<String> getJavaOpts(Configuration conf); static void buildMainArgs(List<String> command,
      String user, String appId, String locId,
      InetSocketAddress nmAddr, List<String> localDirs, Configuration conf); static void main(String[] argv); static final String FILECACHE; static final String APPCACHE; static final String USERCACHE; static final String TOKEN_FILE_NAME_FMT; }
@Test public void testMain() throws Exception { ContainerLocalizerWrapper wrapper = new ContainerLocalizerWrapper(); ContainerLocalizer localizer = wrapper.setupContainerLocalizerForTest(); Random random = wrapper.random; List<Path> localDirs = wrapper.localDirs; Path tokenPath = wrapper.tokenPath; LocalizationProtocol nmProxy = wrapper.nmProxy; AbstractFileSystem spylfs = wrapper.spylfs; mockOutDownloads(localizer); List<Path> privCacheList = new ArrayList<Path>(); List<Path> appCacheList = new ArrayList<Path>(); for (Path p : localDirs) { Path base = new Path(new Path(p, ContainerLocalizer.USERCACHE), appUser); Path privcache = new Path(base, ContainerLocalizer.FILECACHE); privCacheList.add(privcache); Path appDir = new Path(base, new Path(ContainerLocalizer.APPCACHE, appId)); Path appcache = new Path(appDir, ContainerLocalizer.FILECACHE); appCacheList.add(appcache); } ResourceLocalizationSpec rsrcA = getMockRsrc(random, LocalResourceVisibility.PRIVATE, privCacheList.get(0)); ResourceLocalizationSpec rsrcB = getMockRsrc(random, LocalResourceVisibility.PRIVATE, privCacheList.get(0)); ResourceLocalizationSpec rsrcC = getMockRsrc(random, LocalResourceVisibility.APPLICATION, appCacheList.get(0)); ResourceLocalizationSpec rsrcD = getMockRsrc(random, LocalResourceVisibility.PRIVATE, privCacheList.get(0)); when(nmProxy.heartbeat(isA(LocalizerStatus.class))) .thenReturn(new MockLocalizerHeartbeatResponse(LocalizerAction.LIVE, Collections.singletonList(rsrcA))) .thenReturn(new MockLocalizerHeartbeatResponse(LocalizerAction.LIVE, Collections.singletonList(rsrcB))) .thenReturn(new MockLocalizerHeartbeatResponse(LocalizerAction.LIVE, Collections.singletonList(rsrcC))) .thenReturn(new MockLocalizerHeartbeatResponse(LocalizerAction.LIVE, Collections.singletonList(rsrcD))) .thenReturn(new MockLocalizerHeartbeatResponse(LocalizerAction.LIVE, Collections.<ResourceLocalizationSpec>emptyList())) .thenReturn(new MockLocalizerHeartbeatResponse(LocalizerAction.DIE, null)); LocalResource tRsrcA = rsrcA.getResource(); LocalResource tRsrcB = rsrcB.getResource(); LocalResource tRsrcC = rsrcC.getResource(); LocalResource tRsrcD = rsrcD.getResource(); doReturn( new FakeDownload(rsrcA.getResource().getResource().getFile(), true)) .when(localizer).download(isA(Path.class), eq(tRsrcA), isA(UserGroupInformation.class)); doReturn( new FakeDownload(rsrcB.getResource().getResource().getFile(), true)) .when(localizer).download(isA(Path.class), eq(tRsrcB), isA(UserGroupInformation.class)); doReturn( new FakeDownload(rsrcC.getResource().getResource().getFile(), true)) .when(localizer).download(isA(Path.class), eq(tRsrcC), isA(UserGroupInformation.class)); doReturn( new FakeDownload(rsrcD.getResource().getResource().getFile(), true)) .when(localizer).download(isA(Path.class), eq(tRsrcD), isA(UserGroupInformation.class)); localizer.runLocalization(nmAddr); for (Path p : localDirs) { Path base = new Path(new Path(p, ContainerLocalizer.USERCACHE), appUser); Path privcache = new Path(base, ContainerLocalizer.FILECACHE); verify(spylfs).mkdir(eq(privcache), eq(CACHE_DIR_PERM), eq(false)); Path appDir = new Path(base, new Path(ContainerLocalizer.APPCACHE, appId)); Path appcache = new Path(appDir, ContainerLocalizer.FILECACHE); verify(spylfs).mkdir(eq(appcache), eq(CACHE_DIR_PERM), eq(false)); } verify(spylfs).open(tokenPath); verify(nmProxy).heartbeat(argThat(new HBMatches(rsrcA.getResource()))); verify(nmProxy).heartbeat(argThat(new HBMatches(rsrcB.getResource()))); verify(nmProxy).heartbeat(argThat(new HBMatches(rsrcC.getResource()))); verify(nmProxy).heartbeat(argThat(new HBMatches(rsrcD.getResource()))); verify(nmProxy, never()).heartbeat(argThat( new ArgumentMatcher<LocalizerStatus>() { @Override public boolean matches(Object o) { LocalizerStatus status = (LocalizerStatus) o; return !containerId.equals(status.getLocalizerId()); } })); }