Fix AutoPagerAsync hang when first-page handler throws

openai-java-core/src/main/kotlin/com/openai/core/AutoPagerAsync.kt

@@ -45,25 +45,31 @@ private constructor(private val firstPage: PageAsync<T>, private val defaultExec
             else CompletableFuture.completedFuture(null)
         }
 
-        executor.execute {
-            firstPage.handle().whenComplete { _, error ->
-                val actualError =
-                    if (error is CompletionException && error.cause != null) error.cause else error
+        fun complete(error: Throwable?) {
+            val actualError =
+                if (error is CompletionException && error.cause != null) error.cause else error
+            try {
+                handler.onComplete(Optional.ofNullable(actualError))
+            } finally {
                 try {
-                    handler.onComplete(Optional.ofNullable(actualError))
-                } finally {
-                    try {
-                        if (actualError == null) {
-                            onCompleteFuture.complete(null)
-                        } else {
-                            onCompleteFuture.completeExceptionally(actualError)
-                        }
-                    } finally {
-                        close()
+                    if (actualError == null) {
+                        onCompleteFuture.complete(null)
+                    } else {
+                        onCompleteFuture.completeExceptionally(actualError)
                     }
+                } finally {
+                    close()
                 }
             }
         }
+
+        executor.execute {
+            try {
+                firstPage.handle().whenComplete { _, error -> complete(error) }
+            } catch (error: Throwable) {
+                complete(error)
+            }
+        }
     }
 
     override fun onCompleteFuture(): CompletableFuture<Void?> = onCompleteFuture

openai-java-core/src/test/kotlin/com/openai/core/AutoPagerAsyncTest.kt

@@ -14,6 +14,7 @@ import org.mockito.junit.jupiter.MockitoExtension
 import org.mockito.kotlin.any
 import org.mockito.kotlin.clearInvocations
 import org.mockito.kotlin.doAnswer
+import org.mockito.kotlin.doThrow
 import org.mockito.kotlin.inOrder
 import org.mockito.kotlin.mock
 import org.mockito.kotlin.never
@@ -146,6 +147,38 @@ internal class AutoPagerAsyncTest {
         }
     }
 
+    @Test
+    fun subscribe_whenHandlerOnNextThrowsOnFirstPage_callsOnComplete() {
+        val page = PageAsyncImpl(listOf("chunk1", "chunk2"), hasNext = false)
+        val autoPagerAsync = AutoPagerAsync.from(page, executor)
+        doThrow(ERROR).whenever(handler).onNext("chunk1")
+
+        autoPagerAsync.subscribe(handler)
+
+        verify(handler, times(1)).onNext("chunk1")
+        verify(handler, never()).onNext("chunk2")
+        verify(handler, times(1)).onComplete(Optional.of(ERROR))
+        assertThat(autoPagerAsync.onCompleteFuture()).isCompletedExceptionally
+    }
+
+    @Test
+    fun subscribe_whenHandlerOnNextThrowsOnNextPage_callsOnComplete() {
+        val page = PageAsyncImpl(listOf("chunk1"))
+        val autoPagerAsync = AutoPagerAsync.from(page, executor)
+        doThrow(ERROR).whenever(handler).onNext("chunk2")
+
+        autoPagerAsync.subscribe(handler)
+        page.nextPageFuture.complete(PageAsyncImpl(listOf("chunk2", "chunk3"), hasNext = false))
+
+        inOrder(handler) {
+            verify(handler, times(1)).onNext("chunk1")
+            verify(handler, times(1)).onNext("chunk2")
+            verify(handler, times(1)).onComplete(Optional.of(ERROR))
+        }
+        verify(handler, never()).onNext("chunk3")
+        assertThat(autoPagerAsync.onCompleteFuture()).isCompletedExceptionally
+    }
+
     @Test
     fun onCompleteFuture_whenNextPageFutureNotCompleted_onCompleteFutureNotCompleted() {
         val page = PageAsyncImpl(listOf("chunk1", "chunk2"))