### Question:
BackgroundHiveSplitLoader implements HiveSplitLoader { @Override public void start(HiveSplitSource splitSource) { this.hiveSplitSource = splitSource; this.stopwatch = Stopwatch.createStarted(); for (int i = 0; i < loaderConcurrency; i++) { ListenableFuture<?> future = ResumableTasks.submit(executor, new HiveSplitLoaderTask()); addExceptionCallback(future, hiveSplitSource::fail); } } BackgroundHiveSplitLoader(
            Table table,
            Iterable<HivePartitionMetadata> partitions,
            TupleDomain<? extends ColumnHandle> compactEffectivePredicate,
            DynamicFilter dynamicFilter,
            Duration dynamicFilteringProbeBlockingTimeout,
            TypeManager typeManager,
            Optional<BucketSplitInfo> tableBucketInfo,
            ConnectorSession session,
            HdfsEnvironment hdfsEnvironment,
            NamenodeStats namenodeStats,
            DirectoryLister directoryLister,
            Executor executor,
            int loaderConcurrency,
            boolean recursiveDirWalkerEnabled,
            boolean ignoreAbsentPartitions,
            Optional<ValidWriteIdList> validWriteIds); @Override void start(HiveSplitSource splitSource); @Override void stop(); static boolean hasAttemptId(String bucketFilename);  }### Answer:
@Test public void testNoHangIfPartitionIsOffline() { BackgroundHiveSplitLoader backgroundHiveSplitLoader = backgroundHiveSplitLoaderOfflinePartitions(); HiveSplitSource hiveSplitSource = hiveSplitSource(backgroundHiveSplitLoader); backgroundHiveSplitLoader.start(hiveSplitSource); assertThrows(RuntimeException.class, () -> drain(hiveSplitSource)); assertThrows(RuntimeException.class, hiveSplitSource::isFinished); }
@Test(dataProvider = "testPropagateExceptionDataProvider", timeOut = 60_000) public void testPropagateException(boolean error, int threads) { AtomicBoolean iteratorUsedAfterException = new AtomicBoolean(); BackgroundHiveSplitLoader backgroundHiveSplitLoader = new BackgroundHiveSplitLoader( SIMPLE_TABLE, () -> new Iterator<>() { private boolean threw; @Override public boolean hasNext() { iteratorUsedAfterException.compareAndSet(false, threw); return !threw; } @Override public HivePartitionMetadata next() { iteratorUsedAfterException.compareAndSet(false, threw); threw = true; if (error) { throw new Error("loading error occurred"); } throw new RuntimeException("loading error occurred"); } }, TupleDomain.all(), DynamicFilter.EMPTY, Duration.valueOf("0s"), TYPE_MANAGER, createBucketSplitInfo(Optional.empty(), Optional.empty()), SESSION, new TestingHdfsEnvironment(TEST_FILES), new NamenodeStats(), new CachingDirectoryLister(new HiveConfig()), executor, threads, false, false, Optional.empty()); HiveSplitSource hiveSplitSource = hiveSplitSource(backgroundHiveSplitLoader); backgroundHiveSplitLoader.start(hiveSplitSource); assertThatThrownBy(() -> drain(hiveSplitSource)) .hasMessageEndingWith("loading error occurred"); assertThatThrownBy(hiveSplitSource::isFinished) .hasMessageEndingWith("loading error occurred"); if (threads == 1) { assertFalse(iteratorUsedAfterException.get()); } }