HadoopJobLoader { public boolean waitForCompletion() throws Exception { long timestampEpochMillis = System.currentTimeMillis(); String hdfsDir = Constants.HDFS_DATA_DIR + "/" + options.terrapinFileSet + "/" + timestampEpochMillis; ZooKeeperManager zkManager = getZKManager(); int numShards = job.getNumReduceTasks(); FileSetInfo fileSetInfo = new FileSetInfo(options.terrapinFileSet, hdfsDir, numShards, (List) Lists.newArrayList(), options.loadOptions); int replicationFactor = Constants.DEFAULT_HDFS_REPLICATION; String terrapinNamenode = options.terrapinNamenode; if (terrapinNamenode == null || terrapinNamenode.isEmpty()) { ClusterInfo info = zkManager.getClusterInfo(); if (info == null) { LOG.error("Could not cluster information for " + options.terrapinCluster); System.exit(1); } if (info.hdfsNameNode == null || info.hdfsNameNode.isEmpty()) { LOG.error("Could not find the namenode for " + options.terrapinCluster); System.exit(1); } terrapinNamenode = info.hdfsNameNode; replicationFactor = info.hdfsReplicationFactor; } job.setOutputFormatClass(HFileOutputFormat.class); job.setOutputKeyClass(BytesWritable.class); job.setOutputValueClass(BytesWritable.class); setOutputPath(terrapinNamenode, hdfsDir); zkManager.lockFileSet(options.terrapinFileSet, fileSetInfo); try { Configuration conf = job.getConfiguration(); TerrapinUtil.setupConfiguration(conf, Constants.DEFAULT_MAX_SHARD_SIZE_BYTES, replicationFactor); if (!job.waitForCompletion(true)) { LOG.error("MR job failed."); return false; } loadFileSetData(zkManager, fileSetInfo, options.loadOptions); try { LOG.info("Releasing file set lock."); Thread.sleep(5000); } catch (InterruptedException ie) { LOG.warn("Interrupted."); } } finally { zkManager.unlockFileSet(options.terrapinFileSet); } return true; } HadoopJobLoader(TerrapinUploaderOptions options, Job job); boolean waitForCompletion();  }
@Test public void testUpload() throws Exception { ZooKeeperManager zkManager = mock(ZooKeeperManager.class); Job job = mock(Job.class); Configuration conf = new Configuration(); TerrapinUploaderOptions options = new TerrapinUploaderOptions(); options.terrapinFileSet = FILE_SET; when(zkManager.getClusterInfo()).thenReturn(new ClusterInfo(NAME_NODE, REPLICA_FACTOR)); doNothing().when(zkManager).lockFileSet(anyString(), any(FileSetInfo.class)); doNothing().when(zkManager).unlockFileSet(anyString()); when(job.waitForCompletion(anyBoolean())).then(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocationOnMock) throws Throwable { Thread.sleep(1000); return true; } }); when(job.getNumReduceTasks()).thenReturn(NUM_PART); when(job.getConfiguration()).thenReturn(conf); doNothing().when(job).setOutputFormatClass(Matchers.<Class<? extends OutputFormat>>any()); doNothing().when(job).setOutputKeyClass(Matchers.<Class<?>>any()); doNothing().when(job).setOutputValueClass(Matchers.<Class<?>>any()); TestHadoopJobLoader uploader = new TestHadoopJobLoader(zkManager, options, job); assertTrue(uploader.waitForCompletion()); assertEquals(Constants.MAPRED_MAP_MAX_ATTEMPTS, Integer.parseInt(conf.get("mapred.map.max.attempts"))); assertEquals(Constants.CHECKSUM_BYTES, Integer.parseInt(conf.get("io.bytes.per.checksum"))); assertEquals(Constants.DEFAULT_MAX_SHARD_SIZE_BYTES, Long.parseLong(conf.get("dfs.block.size"))); assertEquals(REPLICA_FACTOR, Integer.parseInt(conf.get("dfs.replication"))); verify(zkManager).getClusterInfo(); verify(zkManager).lockFileSet(eq(FILE_SET), any(FileSetInfo.class)); verify(zkManager).unlockFileSet(eq(FILE_SET)); verify(job).waitForCompletion(eq(true)); verify(job).setOutputFormatClass(eq(HFileOutputFormat.class)); verify(job).setOutputValueClass(eq(BytesWritable.class)); verify(job).setOutputKeyClass(eq(BytesWritable.class)); }