KiteLoader extends Loader<LinkConfiguration, ToJobConfiguration> { @Override public void load(LoaderContext context, LinkConfiguration linkConfiguration, ToJobConfiguration toJobConfig) throws Exception { String uri = ConfigUtil.buildDatasetUri( linkConfiguration.linkConfig, toJobConfig.toJobConfig); KiteDatasetExecutor executor = getExecutor( linkConfiguration, uri, context.getSchema(), toJobConfig.toJobConfig.fileFormat); LOG.info("Temporary dataset created."); DataReader reader = context.getDataReader(); Object[] array; boolean success = false; try { while ((array = reader.readArrayRecord()) != null) { executor.writeRecord(array); rowsWritten++; } LOG.info(rowsWritten + " data record(s) have been written into dataset."); success = true; } finally { executor.closeWriter(); if (!success) { LOG.error("Fail to write data, dataset will be removed."); executor.deleteDataset(); } } }  @Override void load(LoaderContext context, LinkConfiguration linkConfiguration,
      ToJobConfiguration toJobConfig); @Override long getRowsWritten();  }
@Test public void testLoader() throws Exception { final int NUMBER_OF_ROWS = 1000; org.apache.sqoop.schema.Schema schema = new org.apache.sqoop.schema.Schema("TestLoader"); schema.addColumn(new Text("TextCol")); DataReader reader = new DataReader() { private long index = 0L; @Override public Object[] readArrayRecord() { if (index++ < NUMBER_OF_ROWS) { return new Object[]{ Long.toString(index), }; } else { return null; } } @Override public String readTextRecord() { return null; } @Override public Object readContent() { return null; } }; LoaderContext context = new LoaderContext(null, reader, schema); LinkConfiguration linkConfig = new LinkConfiguration(); ToJobConfiguration toJobConfig = new ToJobConfiguration(); loader.load(context, linkConfig, toJobConfig); verify(executorMock, times(NUMBER_OF_ROWS)).writeRecord( any(Object[].class)); }