@Test public void testMap() throws Exception { Configuration conf = new Configuration(); conf.set(IMPORT_FACADE_FACTORY_CLASS, "eu.dnetlib.iis.wf.importer.content.MockObjectStoreFacadeFactory"); doReturn(conf).when(context).getConfiguration(); mapper.setup(context); AvroKey<Identifier> key = new AvroKey<>(buildIdentifier("objectStoreId")); mapper.map(key, null, context); int expectedCount = 5; verify(context, times(expectedCount)).write(contentUrlCaptor.capture(), any()); List<DocumentContentUrl> expectedList = loadExpected(); for (int i=0; i < expectedCount; i++) { assertEquals(expectedList.get(i), contentUrlCaptor.getAllValues().get(i).datum()); } }
@Override protected void map(AvroKey<Identifier> key, NullWritable value, Context context) throws IOException, InterruptedException { try { String objectStoreId = key.datum().getId().toString(); long startTime = System.currentTimeMillis(); long intervalTime = startTime; int recordIndex=0; log.info("starting importing process from object store: " + objectStoreId); for (String record : objectStoreFacade.deliverObjects(objectStoreId, 0l, System.currentTimeMillis())) { context.write(new AvroKey<DocumentContentUrl>(buildRecord(record)), NullWritable.get()); if (recordIndex>0 && recordIndex%PROGRESS_LOG_INTERVAL==0) { log.info("content retrieval progress: " + recordIndex + ", time taken to process " + PROGRESS_LOG_INTERVAL + " elements: " + ((System.currentTimeMillis() - intervalTime)/1000) + " secs"); intervalTime = System.currentTimeMillis(); } recordIndex++; } log.info("URL importing process from object store: " + objectStoreId + " has finished"); } catch (ServiceFacadeException e) { throw new IOException("Unable to deliver objects using ObjectStore facade", e); } }
ObjectStoreDocumentContentUrlImporterMapper extends Mapper<AvroKey<Identifier>, NullWritable, AvroKey<DocumentContentUrl>, NullWritable> { @Override protected void map(AvroKey<Identifier> key, NullWritable value, Context context) throws IOException, InterruptedException { try { String objectStoreId = key.datum().getId().toString(); long startTime = System.currentTimeMillis(); long intervalTime = startTime; int recordIndex=0; log.info("starting importing process from object store: " + objectStoreId); for (String record : objectStoreFacade.deliverObjects(objectStoreId, 0l, System.currentTimeMillis())) { context.write(new AvroKey<DocumentContentUrl>(buildRecord(record)), NullWritable.get()); if (recordIndex>0 && recordIndex%PROGRESS_LOG_INTERVAL==0) { log.info("content retrieval progress: " + recordIndex + ", time taken to process " + PROGRESS_LOG_INTERVAL + " elements: " + ((System.currentTimeMillis() - intervalTime)/1000) + " secs"); intervalTime = System.currentTimeMillis(); } recordIndex++; } log.info("URL importing process from object store: " + objectStoreId + " has finished"); } catch (ServiceFacadeException e) { throw new IOException("Unable to deliver objects using ObjectStore facade", e); } } }
ObjectStoreDocumentContentUrlImporterMapper extends Mapper<AvroKey<Identifier>, NullWritable, AvroKey<DocumentContentUrl>, NullWritable> { @Override protected void map(AvroKey<Identifier> key, NullWritable value, Context context) throws IOException, InterruptedException { try { String objectStoreId = key.datum().getId().toString(); long startTime = System.currentTimeMillis(); long intervalTime = startTime; int recordIndex=0; log.info("starting importing process from object store: " + objectStoreId); for (String record : objectStoreFacade.deliverObjects(objectStoreId, 0l, System.currentTimeMillis())) { context.write(new AvroKey<DocumentContentUrl>(buildRecord(record)), NullWritable.get()); if (recordIndex>0 && recordIndex%PROGRESS_LOG_INTERVAL==0) { log.info("content retrieval progress: " + recordIndex + ", time taken to process " + PROGRESS_LOG_INTERVAL + " elements: " + ((System.currentTimeMillis() - intervalTime)/1000) + " secs"); intervalTime = System.currentTimeMillis(); } recordIndex++; } log.info("URL importing process from object store: " + objectStoreId + " has finished"); } catch (ServiceFacadeException e) { throw new IOException("Unable to deliver objects using ObjectStore facade", e); } }  }
ObjectStoreDocumentContentUrlImporterMapper extends Mapper<AvroKey<Identifier>, NullWritable, AvroKey<DocumentContentUrl>, NullWritable> { @Override protected void map(AvroKey<Identifier> key, NullWritable value, Context context) throws IOException, InterruptedException { try { String objectStoreId = key.datum().getId().toString(); long startTime = System.currentTimeMillis(); long intervalTime = startTime; int recordIndex=0; log.info("starting importing process from object store: " + objectStoreId); for (String record : objectStoreFacade.deliverObjects(objectStoreId, 0l, System.currentTimeMillis())) { context.write(new AvroKey<DocumentContentUrl>(buildRecord(record)), NullWritable.get()); if (recordIndex>0 && recordIndex%PROGRESS_LOG_INTERVAL==0) { log.info("content retrieval progress: " + recordIndex + ", time taken to process " + PROGRESS_LOG_INTERVAL + " elements: " + ((System.currentTimeMillis() - intervalTime)/1000) + " secs"); intervalTime = System.currentTimeMillis(); } recordIndex++; } log.info("URL importing process from object store: " + objectStoreId + " has finished"); } catch (ServiceFacadeException e) { throw new IOException("Unable to deliver objects using ObjectStore facade", e); } }   }
ObjectStoreDocumentContentUrlImporterMapper extends Mapper<AvroKey<Identifier>, NullWritable, AvroKey<DocumentContentUrl>, NullWritable> { @Override protected void map(AvroKey<Identifier> key, NullWritable value, Context context) throws IOException, InterruptedException { try { String objectStoreId = key.datum().getId().toString(); long startTime = System.currentTimeMillis(); long intervalTime = startTime; int recordIndex=0; log.info("starting importing process from object store: " + objectStoreId); for (String record : objectStoreFacade.deliverObjects(objectStoreId, 0l, System.currentTimeMillis())) { context.write(new AvroKey<DocumentContentUrl>(buildRecord(record)), NullWritable.get()); if (recordIndex>0 && recordIndex%PROGRESS_LOG_INTERVAL==0) { log.info("content retrieval progress: " + recordIndex + ", time taken to process " + PROGRESS_LOG_INTERVAL + " elements: " + ((System.currentTimeMillis() - intervalTime)/1000) + " secs"); intervalTime = System.currentTimeMillis(); } recordIndex++; } log.info("URL importing process from object store: " + objectStoreId + " has finished"); } catch (ServiceFacadeException e) { throw new IOException("Unable to deliver objects using ObjectStore facade", e); } }    }