HdfsSystemConsumer extends BlockingEnvelopeMap { @Override public void register(SystemStreamPartition systemStreamPartition, String offset) { LOG.info("HdfsSystemConsumer register with partition: " + systemStreamPartition + " and offset " + offset); super.register(systemStreamPartition, offset); MultiFileHdfsReader reader = new MultiFileHdfsReader(readerType, systemStreamPartition, getPartitionDescriptor(systemStreamPartition), offset, numMaxRetires); readers.put(systemStreamPartition, reader); consumerMetrics.registerSystemStreamPartition(systemStreamPartition); } HdfsSystemConsumer(String systemName, Config config, HdfsSystemConsumerMetrics consumerMetrics); @Override void start(); @Override void stop(); @Override void register(SystemStreamPartition systemStreamPartition, String offset); @Override Map<SystemStreamPartition, List<IncomingMessageEnvelope>> poll(
    Set<SystemStreamPartition> systemStreamPartitions, long timeout);  }
@Test public void testEmptyStagingDirectory() throws Exception { Map<String, String> configMap = new HashMap<>(); configMap.put(String.format(HdfsConfig.CONSUMER_PARTITIONER_WHITELIST(), SYSTEM_NAME), ".*avro"); Config config = new MapConfig(configMap); HdfsSystemFactory systemFactory = new HdfsSystemFactory(); HdfsSystemAdmin systemAdmin = systemFactory.getAdmin(SYSTEM_NAME, config); String stream = WORKING_DIRECTORY; Set<String> streamNames = new HashSet<>(); streamNames.add(stream); generateAvroDataFiles(); Map<String, SystemStreamMetadata> streamMetadataMap = systemAdmin.getSystemStreamMetadata(streamNames); SystemStreamMetadata systemStreamMetadata = streamMetadataMap.get(stream); Assert.assertEquals(NUM_FILES, systemStreamMetadata.getSystemStreamPartitionMetadata().size()); HdfsSystemConsumer systemConsumer = systemFactory.getConsumer(SYSTEM_NAME, config, new NoOpMetricsRegistry()); Partition partition = new Partition(0); SystemStreamPartition ssp = new SystemStreamPartition(SYSTEM_NAME, stream, partition); try { systemConsumer.register(ssp, "0"); Assert.fail("Empty staging directory should fail system consumer"); } catch (UncheckedExecutionException e) { Assert.assertTrue(e.getCause() instanceof SamzaException); } }