### Question:
BufferingSplitSource implements SplitSource { @Override public ListenableFuture<SplitBatch> getNextBatch(ConnectorPartitionHandle partitionHandle, Lifespan lifespan, int maxSize) { checkArgument(maxSize > 0, "Cannot fetch a batch of zero size"); return GetNextBatch.fetchNextBatchAsync(source, Math.min(bufferSize, maxSize), maxSize, partitionHandle, lifespan); } BufferingSplitSource(SplitSource source, int bufferSize); @Override CatalogName getCatalogName(); @Override ListenableFuture<SplitBatch> getNextBatch(ConnectorPartitionHandle partitionHandle, Lifespan lifespan, int maxSize); @Override void close(); @Override boolean isFinished(); @Override Optional<Integer> getMinScheduleSplitBatchSize();  }### Answer:
@Test public void testFailImmediate() { MockSplitSource mockSource = new MockSplitSource() .setBatchSize(1) .atSplitCompletion(FAIL); try (SplitSource source = new BufferingSplitSource(mockSource, 100)) { assertFutureFailsWithMockFailure(getNextBatch(source, 200)); assertEquals(mockSource.getNextBatchInvocationCount(), 1); } }
@Test public void testFail() { MockSplitSource mockSource = new MockSplitSource() .setBatchSize(1) .increaseAvailableSplits(1) .atSplitCompletion(FAIL); try (SplitSource source = new BufferingSplitSource(mockSource, 100)) { assertFutureFailsWithMockFailure(getNextBatch(source, 2)); assertEquals(mockSource.getNextBatchInvocationCount(), 2); } }