HiveJobFetchSpout extends BaseRichSpout { @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { this.rmResourceFetcher = new RMResourceFetcher(crawlConfig.endPointConfig.RMBasePaths); this.partitionId = calculatePartitionId(context); if (partitionId < 0 || partitionId > crawlConfig.controlConfig.numTotalPartitions) { throw new IllegalStateException("partitionId should be less than numTotalPartitions with partitionId " + partitionId + " and numTotalPartitions " + crawlConfig.controlConfig.numTotalPartitions); } Class<? extends JobIdPartitioner> partitionerCls = crawlConfig.controlConfig.partitionerCls; try { this.jobFilter = new JobIdFilterByPartition(partitionerCls.newInstance(), crawlConfig.controlConfig.numTotalPartitions, partitionId); } catch (Exception e) { LOG.error("failing instantiating job partitioner class " + partitionerCls.getCanonicalName()); throw new IllegalStateException(e); } this.collector = collector; this.runningJobManager = new RunningJobManager(crawlConfig.zkStateConfig.zkQuorum, crawlConfig.zkStateConfig.zkSessionTimeoutMs, crawlConfig.zkStateConfig.zkRetryTimes, crawlConfig.zkStateConfig.zkRetryInterval, crawlConfig.zkStateConfig.zkRoot, crawlConfig.zkStateConfig.zkLockPath); this.lastFinishAppTime = this.runningJobManager.recoverLastFinishedTime(partitionId); if (this.lastFinishAppTime == 0L) { this.lastFinishAppTime = Calendar.getInstance().getTimeInMillis() - 24 * 60 * 60000L; this.runningJobManager.updateLastFinishTime(partitionId, this.lastFinishAppTime); } } HiveJobFetchSpout(RunningJobCrawlConfig crawlConfig); @Override void open(Map conf, TopologyContext context, SpoutOutputCollector collector); @Override void nextTuple(); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override void ack(Object msgId); @Override void fail(Object msgId);  }
@Test public void testOpen() throws Exception { Map conf = mock(HashMap.class); TopologyContext context = mock(TopologyContext.class); SpoutOutputCollector collector = mock(SpoutOutputCollector.class); when(context.getThisTaskId()).thenReturn(1); when(context.getComponentTasks(anyString())).thenReturn(new ArrayList<Integer>() {{ add(1); }}); HiveJobRunningSourcedStormSpoutProvider provider = new HiveJobRunningSourcedStormSpoutProvider(); HiveJobFetchSpout spout = (HiveJobFetchSpout)provider.getSpout(config, 1); spout.open(conf, context, collector); Long yesterday = Long.valueOf(new String(curator.getData().forPath(SHARE_RESOURCES))); Assert.assertTrue(System.currentTimeMillis() > yesterday); }