ParallelSnapshotReader implements Reader { @Override public List<SourceRecord> poll() throws InterruptedException { List<SourceRecord> allRecords = oldTablesReader.isRunning() ? oldTablesReader.poll() : null; List<SourceRecord> newTablesRecords = newTablesReader.poll(); if (newTablesRecords != null) { if (allRecords == null) { allRecords = newTablesRecords; } else { allRecords.addAll(newTablesRecords); } } else { if (allRecords == null) { completeSuccessfully(); } } return allRecords; } ParallelSnapshotReader(Configuration config,
                                  MySqlTaskContext noSnapshotContext,
                                  Filters snapshotFilters,
                                  MySqlConnectorTask.ServerIdGenerator serverIdGenerator);   ParallelSnapshotReader(BinlogReader oldTablesBinlogReader,
                                                 SnapshotReader newTablesSnapshotReader,
                                                 BinlogReader newTablesBinlogReader); ReconcilingBinlogReader createReconcilingBinlogReader(BinlogReader unifiedReader); @Override void uponCompletion(Runnable handler); @Override void initialize(); @Override void start(); @Override void stop(); @Override State state(); @Override List<SourceRecord> poll(); @Override String name();  }
@Test public void pollReturnsOldIfNewReaderIsStopped() throws InterruptedException { BinlogReader mockOldBinlogReader = mock(BinlogReader.class); SnapshotReader mockNewSnapshotReader = mock(SnapshotReader.class); BinlogReader mockNewBinlogReader = mock(BinlogReader.class); ParallelSnapshotReader parallelSnapshotReader = new ParallelSnapshotReader(mockOldBinlogReader, mockNewSnapshotReader, mockNewBinlogReader); SourceRecord oldBinlogSourceRecord = mock(SourceRecord.class); List<SourceRecord> oldBinlogRecords = new ArrayList<>(); oldBinlogRecords.add(oldBinlogSourceRecord); when(mockOldBinlogReader.isRunning()).thenReturn(true); when(mockOldBinlogReader.poll()).thenReturn(oldBinlogRecords); List<SourceRecord> parallelRecords = parallelSnapshotReader.poll(); Assert.assertEquals(1, parallelRecords.size()); Assert.assertTrue(parallelRecords.contains(oldBinlogSourceRecord)); }
@Test public void pollReturnsNullIfBothReadersAreStopped() throws InterruptedException { BinlogReader mockOldBinlogReader = mock(BinlogReader.class); SnapshotReader mockNewSnapshotReader = mock(SnapshotReader.class); BinlogReader mockNewBinlogReader = mock(BinlogReader.class); ParallelSnapshotReader parallelSnapshotReader = new ParallelSnapshotReader(mockOldBinlogReader, mockNewSnapshotReader, mockNewBinlogReader); when(mockOldBinlogReader.isRunning()).thenReturn(false); when(mockOldBinlogReader.poll()).thenThrow(new InterruptedException()); when(mockNewBinlogReader.poll()).thenReturn(null); List<SourceRecord> parallelRecords = parallelSnapshotReader.poll(); Assert.assertEquals(null, parallelRecords); }