SqoopLoader { public RecordWriter<Text, NullWritable> getRecordWriter() { consumerFuture = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat ("OutputFormatLoader-consumer").build()).submit( new ConsumerThread(context)); return writer; } SqoopLoader(JobContext jobctx, String loaderName, IntermediateDataFormat<?> toDataFormat, Matcher matcher);  SqoopLoader(JobContext jobctx); RecordWriter<Text, NullWritable> getRecordWriter(); static final Logger LOG; }
@Test(expectedExceptions = BrokenBarrierException.class) public void testWhenLoaderThrows() throws Throwable { conf.set(MRJobConstants.JOB_ETL_LOADER, ThrowingLoader.class.getName()); SqoopLoader executor = new SqoopLoader(jobContextMock, ThrowingLoader.class.getName(), getIDF(), getMatcher()); RecordWriter<Text, NullWritable> writer = executor .getRecordWriter(); IntermediateDataFormat<?> dataFormat = MRJobTestUtil.getTestIDF(); try { for (int count = 0; count < 100; count++) { dataFormat.setCSVTextData(String.valueOf(count)); writer.write(new Text(dataFormat.getCSVTextData()), null); } } catch (SqoopException ex) { throw ex.getCause(); } }
@Test public void testSuccessfulContinuousLoader() throws Throwable { conf.set(MRJobConstants.JOB_ETL_LOADER, GoodContinuousLoader.class.getName()); SqoopLoader executor = new SqoopLoader(jobContextMock, GoodContinuousLoader.class.getName(), getIDF(), getMatcher()); RecordWriter<Text, NullWritable> writer = executor.getRecordWriter(); IntermediateDataFormat<?> dataFormat = MRJobTestUtil.getTestIDF(); for (int i = 0; i < 10; i++) { StringBuilder builder = new StringBuilder(); for (int count = 0; count < 100; count++) { builder.append(String.valueOf(count)); if (count != 99) { builder.append(","); } } dataFormat.setCSVTextData(builder.toString()); writer.write(new Text(dataFormat.getCSVTextData()), null); } writer.close(null); verify(jobContextMock, times(1)).getConfiguration(); verify(jobContextMock, times(1)).getCounter(SqoopCounters.ROWS_WRITTEN); }
@Test(expectedExceptions = SqoopException.class) public void testSuccessfulLoader() throws Throwable { SqoopLoader executor = new SqoopLoader(jobContextMock, GoodLoader.class.getName(), getIDF(), getMatcher()); RecordWriter<Text, NullWritable> writer = executor .getRecordWriter(); IntermediateDataFormat<?> dataFormat = MRJobTestUtil.getTestIDF(); StringBuilder builder = new StringBuilder(); for (int count = 0; count < 100; count++) { builder.append(String.valueOf(count)); if (count != 99) { builder.append(","); } } dataFormat.setCSVTextData(builder.toString()); writer.write(new Text(dataFormat.getCSVTextData()), null); TimeUnit.SECONDS.sleep(5); writer.close(null); verify(jobContextMock, times(1)).getConfiguration(); verify(jobContextMock, times(1)).getCounter(SqoopCounters.ROWS_WRITTEN); }
@Test(expectedExceptions = ConcurrentModificationException.class) public void testThrowingContinuousLoader() throws Throwable { conf.set(MRJobConstants.JOB_ETL_LOADER, ThrowingContinuousLoader.class.getName()); SqoopLoader executor = new SqoopLoader(jobContextMock, ThrowingContinuousLoader.class.getName(), getIDF(), getMatcher()); RecordWriter<Text, NullWritable> writer = executor.getRecordWriter(); IntermediateDataFormat<?> dataFormat = MRJobTestUtil.getTestIDF(); try { for (int i = 0; i < 10; i++) { StringBuilder builder = new StringBuilder(); for (int count = 0; count < 100; count++) { builder.append(String.valueOf(count)); if (count != 99) { builder.append(","); } } dataFormat.setCSVTextData(builder.toString()); writer.write(new Text(dataFormat.getCSVTextData()), null); } writer.close(null); } catch (SqoopException ex) { throw ex.getCause(); } }