This is an automated email from the ASF dual-hosted git repository.
rustyrazorblade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cassandra-easy-stress.git
The following commit(s) were added to refs/heads/main by this push:
new 40e25eb Fix pagination (#107)
40e25eb is described below
commit 40e25eb796308c6e0cafae4ea96170d05e670d08
Author: Jon Haddad <[email protected]>
AuthorDate: Tue Oct 13 19:32:38 2026 -0700
Fix pagination (#107)
* Fix --paginate looping forever on the first page
In driver v4, AsyncResultSet.fetchNextPage() returns a new page and
does not advance the current one. The callback called it in a loop on
the first page, so hasMorePages() never became false. Each pass sent
another request for page 2, and the driver I/O thread never returned.
Chain the page fetches instead, and record the timer and the result
after the last page arrives. A failed page fetch is now an error.
* Move pagination out of OperationCallback
Fetch the remaining pages where the request future is built, with an
AsyncResultSet.fetchAllPages() extension, instead of inside the result
callback. OperationCallback goes back to its single result/error path
and loses the paginate flag; a failed page fetch reaches the existing
error branch.
The callback now gets the last page, so the first page is not held for
the whole fetch. The DDL special case and the mutable paginate flag are
gone: a single-page result passes straight through.
Reviewed by Nate McCall
---
.../cassandra/easystress/OperationCallback.kt | 8 ---
.../org/apache/cassandra/easystress/Pagination.kt | 33 ++++++++++
.../apache/cassandra/easystress/WorkloadRunner.kt | 26 ++------
.../apache/cassandra/easystress/PaginationTest.kt | 77 ++++++++++++++++++++++
4 files changed, 116 insertions(+), 28 deletions(-)
diff --git
a/src/main/kotlin/org/apache/cassandra/easystress/OperationCallback.kt
b/src/main/kotlin/org/apache/cassandra/easystress/OperationCallback.kt
index c61d5c2..b41eab0 100644
--- a/src/main/kotlin/org/apache/cassandra/easystress/OperationCallback.kt
+++ b/src/main/kotlin/org/apache/cassandra/easystress/OperationCallback.kt
@@ -36,7 +36,6 @@ class OperationCallback(
val op: Operation,
val startNanos: Long,
val populatePhase: Boolean,
- val paginate: Boolean = false,
) : BiConsumer<AsyncResultSet?, Throwable?> {
companion object {
val log = logger()
@@ -53,13 +52,6 @@ class OperationCallback(
return
}
- // Handle pagination in driver v4
- if (paginate && result != null) {
- // Fetch next page - this could be made async but we'll keep it
simple for now
- while (result.hasMorePages()) {
- result.fetchNextPage()
- }
- }
val endNanos = System.nanoTime()
context.timer(op, populatePhase).update(endNanos - op.createdAtNanos,
TimeUnit.NANOSECONDS)
// TODO (visibility): include details about paging?
diff --git a/src/main/kotlin/org/apache/cassandra/easystress/Pagination.kt
b/src/main/kotlin/org/apache/cassandra/easystress/Pagination.kt
new file mode 100644
index 0000000..a4c7b36
--- /dev/null
+++ b/src/main/kotlin/org/apache/cassandra/easystress/Pagination.kt
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.easystress
+
+import com.datastax.oss.driver.api.core.cql.AsyncResultSet
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.CompletionStage
+
+/**
+ * Fetches every remaining page and completes with the last one.
+ * Each page is a separate AsyncResultSet; fetchNextPage() does not advance
the receiver.
+ */
+fun AsyncResultSet.fetchAllPages(): CompletionStage<AsyncResultSet> =
+ if (hasMorePages()) {
+ fetchNextPage().thenCompose { it.fetchAllPages() }
+ } else {
+ CompletableFuture.completedFuture(this)
+ }
diff --git a/src/main/kotlin/org/apache/cassandra/easystress/WorkloadRunner.kt
b/src/main/kotlin/org/apache/cassandra/easystress/WorkloadRunner.kt
index 0272403..5ead439 100644
--- a/src/main/kotlin/org/apache/cassandra/easystress/WorkloadRunner.kt
+++ b/src/main/kotlin/org/apache/cassandra/easystress/WorkloadRunner.kt
@@ -146,14 +146,13 @@ class WorkloadRunner(
// pull requests off the queue instead of using generateKey
// move the getNextOperation into the queue thing
- var paginate = context.mainArguments.paginate
+ val paginate = context.mainArguments.paginate
for (op in queue.getNextOperation()) {
// In driver v4, async execution returns a CompletionStage
val startNanos = System.nanoTime()
val future =
when (op) {
is Operation.DDL -> {
- paginate = false
// Create a SimpleStatement for DDL operations
context.session.executeAsync(SimpleStatement.newInstance(op.statement!!))
}
@@ -163,20 +162,10 @@ class WorkloadRunner(
}
}
- // Create callback to handle the result
- val callback =
- OperationCallback(
- context,
- runner,
- op,
- startNanos,
- queue.populatePhase,
- paginate = paginate,
- )
-
- future.whenComplete { result, error ->
- callback.accept(result, error)
- }
+ // with pagination the operation completes after the last page
arrives
+ val completed = if (paginate) future.thenCompose {
it.fetchAllPages() } else future
+
+ completed.whenComplete(OperationCallback(context, runner, op,
startNanos, queue.populatePhase))
}
}
@@ -220,12 +209,9 @@ class WorkloadRunner(
op,
startNanos,
queue.populatePhase,
- paginate = false,
)
- future.whenComplete { result, error ->
- callback.accept(result, error)
- }
+ future.whenComplete(callback)
}
} catch (_: OperationStopException) {
log.info("Received Stop signal")
diff --git a/src/test/kotlin/org/apache/cassandra/easystress/PaginationTest.kt
b/src/test/kotlin/org/apache/cassandra/easystress/PaginationTest.kt
new file mode 100644
index 0000000..15e5505
--- /dev/null
+++ b/src/test/kotlin/org/apache/cassandra/easystress/PaginationTest.kt
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.easystress
+
+import com.datastax.oss.driver.api.core.cql.AsyncResultSet
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.catchThrowable
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.Timeout
+import java.util.concurrent.CompletableFuture
+
+// a pagination hang must fail the test rather than block the build
+@Timeout(value = 5, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
+class PaginationTest {
+ private fun page(next: CompletableFuture<AsyncResultSet>?): AsyncResultSet
=
+ mockk {
+ every { hasMorePages() } returns (next != null)
+ if (next != null) {
+ every { fetchNextPage() } returns next
+ }
+ }
+
+ @Test
+ fun `fetches each page once and completes with the last page`() {
+ val thirdFuture = CompletableFuture<AsyncResultSet>()
+ val third = page(null)
+ val second = page(thirdFuture)
+ val first = page(CompletableFuture.completedFuture(second))
+
+ val all = first.fetchAllPages().toCompletableFuture()
+
+ verify(exactly = 1) { first.fetchNextPage() }
+ verify(exactly = 1) { second.fetchNextPage() }
+ // the last page has not arrived yet
+ assertThat(all).isNotDone
+
+ thirdFuture.complete(third)
+
+ assertThat(all).isCompletedWithValue(third)
+ }
+
+ @Test
+ fun `completes a single page without fetching`() {
+ val only = page(null)
+
+
assertThat(only.fetchAllPages().toCompletableFuture()).isCompletedWithValue(only)
+ }
+
+ @Test
+ fun `fails when a page fetch fails`() {
+ val failure = RuntimeException("page fetch failed")
+ val first = page(CompletableFuture.failedFuture(failure))
+
+ val all = first.fetchAllPages().toCompletableFuture()
+
+ assertThat(all).isCompletedExceptionally
+ assertThat(catchThrowable { all.join() }.cause).isSameAs(failure)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
