ParentTaskDataFetcher extends DataFetcher { @Override Object fetchDataElement() throws IOException { try { if (firstFetch) { fetchDataLazily(); advanceIterator(); firstFetch = false; } while (true) { if (this.currentIterator.hasNext()) { return this.currentIterator.next(); } if (currentIteratorIndex < expectedNumOfIterators) { countBytes(currentIterator); advanceIterator(); continue; } else { break; } } } catch (final Throwable e) { throw new IOException(e); } return Finishmark.getInstance(); } ParentTaskDataFetcher(final IRVertex dataSource,
                        final InputReader inputReader,
                        final OutputCollector outputCollector); @Override void close();  }
@Test(timeout = 5000) public void testEmpty() throws Exception { final List<String> empty = new ArrayList<>(0); final InputReader inputReader = generateInputReader(generateCompletableFuture(empty.iterator())); final ParentTaskDataFetcher fetcher = createFetcher(inputReader); assertEquals(Finishmark.getInstance(), fetcher.fetchDataElement()); }
@Test(timeout = 5000) public void testNull() throws Exception { final List<String> oneNull = new ArrayList<>(1); oneNull.add(null); final InputReader inputReader = generateInputReader(generateCompletableFuture(oneNull.iterator())); final ParentTaskDataFetcher fetcher = createFetcher(inputReader); assertEquals(null, fetcher.fetchDataElement()); }
@Test(timeout = 5000) public void testNonEmpty() throws Exception { final String singleData = "Single"; final List<String> dataElements = new ArrayList<>(1); dataElements.add(singleData); final InputReader inputReader = generateInputReader(generateCompletableFuture(dataElements.iterator())); final ParentTaskDataFetcher fetcher = createFetcher(inputReader); assertEquals(singleData, fetcher.fetchDataElement()); }
@Test(timeout = 5000, expected = IOException.class) public void testErrorWhenFuture() throws Exception { final CompletableFuture failingFuture = generateFailingFuture(); final InputReader inputReader = generateInputReader(failingFuture); final ParentTaskDataFetcher fetcher = createFetcher(inputReader); fetcher.fetchDataElement(); assertTrue(failingFuture.isCompletedExceptionally()); }
@Test(timeout = 5000) public void testErrorWhenFutureWithRetry() throws Exception { final CompletableFuture failingFuture = generateFailingFuture(); final InputReader inputReader = generateInputReader( failingFuture, BlockFetchFailureProperty.of(BlockFetchFailureProperty.Value.RETRY_AFTER_TWO_SECONDS_FOREVER)); final List<String> empty = new ArrayList<>(0); when(inputReader.retry(anyInt())) .thenReturn(generateCompletableFuture( empty.iterator())); final ParentTaskDataFetcher fetcher = createFetcher(inputReader); assertEquals(Finishmark.getInstance(), fetcher.fetchDataElement()); }
@Test(timeout = 5000, expected = IOException.class) public void testErrorWhenReadingData() throws Exception { final InputReader inputReader = generateInputReader(generateCompletableFuture(new FailedIterator())); final ParentTaskDataFetcher fetcher = createFetcher(inputReader); fetcher.fetchDataElement(); }