### Question:
HiveSplitSource implements ConnectorSplitSource { void fail(Throwable e) { if (setIf(stateReference, State.failed(e), state -> state.getKind() == INITIAL)) { splitLoader.stop(); queues.finish(); } } private  HiveSplitSource(
            ConnectorSession session,
            String databaseName,
            String tableName,
            PerBucket queues,
            int maxInitialSplits,
            DataSize maxOutstandingSplitsSize,
            HiveSplitLoader splitLoader,
            AtomicReference<State> stateReference,
            CounterStat highMemorySplitSourceCounter); static HiveSplitSource allAtOnce(
            ConnectorSession session,
            String databaseName,
            String tableName,
            int maxInitialSplits,
            int maxOutstandingSplits,
            DataSize maxOutstandingSplitsSize,
            int maxSplitsPerSecond,
            HiveSplitLoader splitLoader,
            Executor executor,
            CounterStat highMemorySplitSourceCounter); static HiveSplitSource bucketed(
            ConnectorSession session,
            String databaseName,
            String tableName,
            int estimatedOutstandingSplitsPerBucket,
            int maxInitialSplits,
            DataSize maxOutstandingSplitsSize,
            int maxSplitsPerSecond,
            HiveSplitLoader splitLoader,
            Executor executor,
            CounterStat highMemorySplitSourceCounter); @Override CompletableFuture<ConnectorSplitBatch> getNextBatch(ConnectorPartitionHandle partitionHandle, int maxSize); @Override boolean isFinished(); @Override void close();  }### Answer:
@Test public void testFail() { HiveSplitSource hiveSplitSource = HiveSplitSource.allAtOnce( SESSION, "database", "table", 10, 10, DataSize.of(1, MEGABYTE), Integer.MAX_VALUE, new TestingHiveSplitLoader(), Executors.newFixedThreadPool(5), new CounterStat()); for (int i = 0; i < 5; i++) { hiveSplitSource.addToQueue(new TestSplit(i)); assertEquals(hiveSplitSource.getBufferedInternalSplitCount(), i + 1); } assertEquals(getSplits(hiveSplitSource, 1).size(), 1); assertEquals(hiveSplitSource.getBufferedInternalSplitCount(), 4); hiveSplitSource.fail(new RuntimeException("test")); assertEquals(hiveSplitSource.getBufferedInternalSplitCount(), 4); try { getSplits(hiveSplitSource, 1); fail("expected RuntimeException"); } catch (RuntimeException e) { assertEquals(e.getMessage(), "test"); } assertEquals(hiveSplitSource.getBufferedInternalSplitCount(), 4); hiveSplitSource.addToQueue(new TestSplit(99)); assertEquals(hiveSplitSource.getBufferedInternalSplitCount(), 4); hiveSplitSource.fail(new RuntimeException("another failure")); assertEquals(hiveSplitSource.getBufferedInternalSplitCount(), 4); try { getSplits(hiveSplitSource, 1); fail("expected RuntimeException"); } catch (RuntimeException e) { assertEquals(e.getMessage(), "test"); } }