Async operations

Applies to

• Aerospike Developer SDK preview (Java 21+ and Python 3.10+)
• Aerospike Database 6.0 or later unless a section states otherwise

Code examples

Examples assume a connected session from Connect to Aerospike. Each usage guide lists shared imports in a tabbed block near the top of the page; snippet blocks add imports only for types not already shown. The first code block in each section includes dataset setup; later blocks in that section reuse those values. When a page includes a Complete example section, that block is fully self-contained.

Learn how to use asynchronous operations for high-throughput, non-blocking database access. Async operations allow your application to continue processing while waiting for database responses.

Except where noted, snippets on this page use the imports below. A snippet lists additional import lines only when it needs a type not shown here. When this page includes a Complete example section, that block is fully self-contained with every import required to run it.

Java

import com.aerospike.client.sdk.DataSet;
import com.aerospike.client.sdk.ErrorStrategy;
import com.aerospike.client.sdk.Record;
import com.aerospike.client.sdk.RecordResult;
import com.aerospike.client.sdk.RecordStream;

import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.function.Function;

Python

import asyncio

from aerospike_sdk import DataSet, ErrorStrategy

Why async?

Approach	Behavior	Best For
Sync	Blocks thread until response	Simple scripts, low concurrency
Async	Returns immediately, notifies on completion	High throughput, web servers, microservices

Note

The Java SDK uses virtual threads and hence requires JDK 21 or above. However, performance tests have shown that JDK 26 offers significant performance benefits for virtual threads in high performance environments. For these environments, JDK 26 or above is recommended.

See Error handling for more information about async execution modes.

Async read

Java

DataSet users = DataSet.of("test", "users");

// Non-blocking: results are published to the returned stream as they arrive.
RecordStream stream = session.query(users.id("user-1")).executeAsync(ErrorStrategy.IN_STREAM);
stream.getFirst().ifPresent(result -> {
    if (result.isOk()) {
        Record user = result.recordOrThrow();
        System.out.println("Name: " + user.getString("name"));
    }
});

📖 API reference: DataSet.of(...) | DataSet.id(...) | Session.query(Key) | ChainableQueryBuilder.executeAsync(...) | RecordStream.getFirst() | RecordResult.isOk() | RecordResult.recordOrThrow() | Record.getString(...) | ErrorStrategy | ErrorStrategy.IN_STREAM

Python

users = DataSet.of("test", "users")

# Non-blocking: results are published to the returned stream as they arrive.
stream = await session.query(users.id("user-1")).execute(on_error=ErrorStrategy.IN_STREAM)
row = await stream.first()
if row is not None and row.is_ok:
    user = row.record_or_raise()
    print(f"Name: {user.bins['name']}")

📖 API reference: DataSet.of() | DataSet.id() | Session.query() | QueryBuilder.execute() | ErrorStrategy.IN_STREAM | RecordStream.first() | RecordResult.is_ok | RecordResult.record_or_raise()

Async write

Java

DataSet users = DataSet.of("test", "users");

RecordStream writeStream = session.insert(users)
    .bins("name", "email")
    .id("user-1").values("Alice", "alice@example.com")
    .executeAsync(ErrorStrategy.IN_STREAM);
writeStream.forEach(result -> {
    if (result.isOk()) {
        System.out.println("Insert complete for key index " + result.index());
    }
});

📖 API reference: DataSet.of(...) | Session.insert(DataSet) | OperationObjectBuilder.bins(...) | IdValuesBuilder.id(...) | IdValuesRowBuilder.values(...) | ChainableQueryBuilder.executeAsync(...) | RecordStream.forEach(...) | RecordResult.isOk() | ErrorStrategy | ErrorStrategy.IN_STREAM

Python

users = DataSet.of("test", "users")

write_stream = await (
    session.batch()
    .insert(users.id("user-1")).put({"name": "Alice", "email": "alice@example.com"})
    .execute(on_error=ErrorStrategy.IN_STREAM)
)
async for row in write_stream:
    if row.is_ok:
        print(f"Insert complete for key index {row.index}")
write_stream.close()

📖 API reference: DataSet.of() | DataSet.id() | Session.batch() | BatchOperationBuilder.insert() | BatchKeyOperationBuilder.put() | BatchOperationBuilder.execute() | ErrorStrategy | ErrorStrategy.IN_STREAM | RecordResult.is_ok | RecordResult.index | RecordStream.close()

Parallel operations

Execute multiple operations concurrently:

Java

DataSet users = DataSet.of("test", "users");

// Parallelize with virtual threads: each task runs a synchronous query and maps the first record.
Function<String, Optional<Record>> readOne = id -> {
    RecordStream s = session.query(users.id(id)).execute();
    return s.getFirst().filter(RecordResult::isOk).map(RecordResult::recordOrThrow);
};

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    CompletableFuture<Optional<Record>> u1 = CompletableFuture.supplyAsync(() -> readOne.apply("user-1"), executor);
    CompletableFuture<Optional<Record>> u2 = CompletableFuture.supplyAsync(() -> readOne.apply("user-2"), executor);
    CompletableFuture<Optional<Record>> u3 = CompletableFuture.supplyAsync(() -> readOne.apply("user-3"), executor);
    CompletableFuture.allOf(u1, u2, u3).join();
    List<Optional<Record>> results = List.of(u1.join(), u2.join(), u3.join());
}

📖 API reference: DataSet.of(...) | Session.query(Key) | ChainableQueryBuilder.execute() | RecordStream.getFirst() | RecordResult.isOk()

Python

users = DataSet.of("test", "users")

async def read_one(uid: str):
    stream = await session.query(users.id(uid)).execute(on_error=ErrorStrategy.IN_STREAM)
    row = await stream.first()
    return row.record_or_raise() if row and row.is_ok else None

# Run three reads concurrently on the asyncio event loop.
results = await asyncio.gather(
    read_one("user-1"),
    read_one("user-2"),
    read_one("user-3"),
)

📖 API reference: DataSet.of() | DataSet.id() | Session.query() | QueryBuilder.execute() | ErrorStrategy.IN_STREAM | RecordStream.first() | RecordResult.is_ok | RecordResult.record_or_raise()

Async with callbacks

Handle results with callbacks (Java) or an await chain (Python) instead of blocking:

Java

DataSet users = DataSet.of("test", "users");

CompletableFuture.runAsync(() -> {
    RecordStream stream = session.query(users.id("user-1")).executeAsync(ErrorStrategy.IN_STREAM);
    stream.getFirst().ifPresent(result -> {
        if (result.isOk()) {
            Record user = result.recordOrThrow();
            System.out.println("Got user: " + user.getString("name"));
        }
    });
});

System.out.println("Operation submitted");

📖 API reference: DataSet.of(...) | DataSet.id(...) | Session.query(Key) | ChainableQueryBuilder.executeAsync(...) | RecordStream.getFirst() | RecordResult.isOk() | RecordResult.recordOrThrow() | Record.getString(...) | ErrorStrategy | ErrorStrategy.IN_STREAM

Python

users = DataSet.of("test", "users")

async def fetch_user():
    stream = await session.query(users.id("user-1")).execute(on_error=ErrorStrategy.IN_STREAM)
    row = await stream.first()
    if row is not None and row.is_ok:
        user = row.record_or_raise()
        print(f"Got user: {user.bins['name']}")

asyncio.create_task(fetch_user())

print("Operation submitted")

📖 API reference: DataSet.of() | DataSet.id() | Session.query() | QueryBuilder.execute() | ErrorStrategy.IN_STREAM | RecordStream.first() | RecordResult.is_ok | RecordResult.record_or_raise()

Async pipeline

Chain operations together:

Java

DataSet users = DataSet.of("test", "users");
DataSet orders = DataSet.of("test", "orders");
session.query(users.id("user-1"))
    .bins("loyalty_tier")
    .executeAsync(ErrorStrategy.IN_STREAM)
    .asCompletableFuture()
    .thenCompose(results -> {
        String tier = results.get(0).recordOrThrow().getString("loyalty_tier");
        double discount = "gold".equals(tier) ? 0.20 : 0.05;
        return session.update(orders.id("order-99"))
            .bin("discount").setTo(discount)
            .executeAsync(ErrorStrategy.IN_STREAM)
            .asCompletableFuture();
    })
    .thenRun(() -> System.out.println("Discount applied!"))
    .exceptionally(ex -> {
        System.err.println("Pipeline failed: " + ex.getMessage());
        return null;
    });

📖 API reference: DataSet.of(...) | DataSet.id(...) | Session.update(DataSet) | Session.query(Key) | OperationObjectBuilder.bins(...) | ChainableOperationBuilder.bin(...) | ChainableQueryBuilder.bin(...) | ChainableQueryBuilder.executeAsync(...) | RecordStream.asCompletableFuture() | RecordResult.recordOrThrow() | Record.getString(...) | ErrorStrategy | ErrorStrategy.IN_STREAM

Python

users = DataSet.of("test", "users")
orders = DataSet.of("test", "orders")

try:
    read_stream = await (
        session.query(users.id("user-1"))
        .bins(["loyalty_tier"])
        .execute(on_error=ErrorStrategy.IN_STREAM)
    )
    row = await read_stream.first()
    tier = row.record_or_raise().bins["loyalty_tier"]
    discount = 0.20 if tier == "gold" else 0.05
    await (
        session.update(orders.id("order-99"))
        .bin("discount").set_to(discount)
        .execute(on_error=ErrorStrategy.IN_STREAM)
    )
    print("Discount applied!")
except Exception as ex:
    print(f"Pipeline failed: {ex}")

📖 API reference: DataSet.of() | DataSet.id() | Session.query() | Session.update() | QueryBuilder.bins() | QueryBuilder.execute() | ErrorStrategy.IN_STREAM | RecordStream.first() | RecordResult.record_or_raise() | WriteSegmentBuilder.set_to() | WriteSegmentBuilder.bin() | WriteSegmentBuilder.execute()

Async batch

Batch operations also support async:

Java

DataSet users = DataSet.of("test", "users");

CompletableFuture.runAsync(() -> {
    RecordStream stream = session
        .query(users.id("user-1"), users.id("user-2"), users.id("user-3"))
        .executeAsync(ErrorStrategy.IN_STREAM);
    long count = 0;
    try {
        while (stream.hasNext()) {
            RecordResult rr = stream.next();
            if (rr.isOk() && rr.recordOrNull() != null) {
                count++;
            }
        }
    } finally {
        stream.close();
    }
    System.out.println("Batch complete, got " + count + " records");
});

📖 API reference: DataSet.of(...) | DataSet.id(...) | ChainableQueryBuilder.executeAsync(...) | RecordStream.hasNext() | RecordStream.next() | RecordStream.close() | RecordResult.isOk() | ErrorStrategy | ErrorStrategy.IN_STREAM

Python

users = DataSet.of("test", "users")

stream = await (
    session.query(users.ids("user-1", "user-2", "user-3"))
    .execute(on_error=ErrorStrategy.IN_STREAM)
)
count = 0
try:
    async for rr in stream:
        if rr.is_ok and rr.record is not None:
            count += 1
finally:
    stream.close()
print(f"Batch complete, got {count} records")

📖 API reference: DataSet.of() | DataSet.ids() | Session.query() | QueryBuilder.execute() | ErrorStrategy.IN_STREAM | RecordResult.is_ok | RecordStream.close()

Async query

Stream query results asynchronously:

Java

DataSet users = DataSet.of("test", "users");

CompletableFuture.runAsync(() -> {
    RecordStream queryStream = session.query(users)
        .where("$.status == 'active'")
        .executeAsync(ErrorStrategy.IN_STREAM);
    queryStream.forEach((RecordResult result) -> {
        if (result.isOk()) {
            Record user = result.recordOrThrow();
            System.out.println("Active user: " + user.getString("name"));
        }
    });
});

📖 API reference: DataSet.of(...) | Session.query(DataSet) | ChainableQueryBuilder.where(...) | ChainableQueryBuilder.executeAsync(...) | RecordStream.forEach(...) | RecordResult.isOk() | RecordResult.recordOrThrow() | Record.getString(...) | ErrorStrategy | ErrorStrategy.IN_STREAM

Python

users = DataSet.of("test", "users")

stream = await (
    session.query(users)
    .where("$.status == 'active'")
    .execute(on_error=ErrorStrategy.IN_STREAM)
)

async for row in stream:
    if row.is_ok:
        print(f"Active user: {row.record_or_raise().bins['name']}")
stream.close()

📖 API reference: DataSet.of() | Session.query() | QueryBuilder.where() | QueryBuilder.execute() | ErrorStrategy.IN_STREAM | RecordResult.is_ok | RecordResult.record_or_raise() | RecordStream.close()

AsyncPool (Python, free-threaded builds)

AsyncPool runs your async work on several CPU cores at once for higher throughput. Each pool loop gets its own Client — the pool calls client_factory() once per loop at construction, so a pool with loop_count=4 holds 4 Clients total and reuses them across every dispatched item.

It only speeds things up on free-threaded CPython (run with PYTHON_GIL=0); on standard CPython it still works correctly but gives no speedup, because the GIL forces all the work onto one core.

# Additional imports for this example:
from aerospike_sdk import AsyncPool, Behavior, Client


async def main():
    pool = AsyncPool(
        client_factory=lambda: Client("localhost:3000"),
        loop_count=4,
    )
    async with pool:
        users = DataSet.of("test", "users")

        # One-shot dispatch via pool.run:
        await pool.run(
            lambda client: client.create_session(Behavior.DEFAULT)
            .upsert(users.id(1))
            .put({"name": "Alice"})
            .execute()
        )

        # Run 1000 upserts in parallel, spread across the pool's 4 loops:
        async def upsert_one(client: Client, uid: int) -> None:
            await (
                client.create_session(Behavior.DEFAULT)
                .upsert(users.id(uid))
                .put({"name": f"User {uid}"})
                .execute()
            )

        await pool.map(upsert_one, range(1, 1001))

For fast-path single-key reads and writes (session.get, session.put), performance mode trade-offs, and benchmark guidance, see the Python SDK performance guide.

Java reactive streams with asPublisher()

asPublisher() returns a java.util.concurrent.Flow.Publisher<RecordResult> with backpressure. The stream closes automatically when the subscription completes, errors, or is cancelled. This is unicast only; a second subscriber receives onError(IllegalStateException).

Use it for large or unbounded query results where collecting everything via asCompletableFuture() is impractical:

// Additional imports for this example:
import java.util.concurrent.Flow;

DataSet users = DataSet.of("test", "users");

Flow.Publisher<RecordResult> publisher = session.query(users)
    .where("$.status == 'active'")
    .executeAsync(ErrorStrategy.IN_STREAM)
    .asPublisher();

publisher.subscribe(new Flow.Subscriber<>() {
    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription sub) {
        subscription = sub;
        sub.request(100);  // initial batch; request more in onNext for backpressure
    }

    @Override
    public void onNext(RecordResult result) {
        if (result.isOk()) {
            System.out.println(result.recordOrThrow().getString("name"));
        }
        subscription.request(1);
    }

    @Override
    public void onError(Throwable t) {
        System.err.println("Publisher error: " + t.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("Query complete");
    }
});

📖 API reference: Session.query(DataSet) | ChainableQueryBuilder.where(...) | ChainableQueryBuilder.executeAsync(...) | RecordStream.asPublisher() | RecordResult.isOk() | RecordResult.recordOrThrow() | ErrorStrategy.IN_STREAM

Async error handling

Async operations use the same error modes as synchronous execute(). Pass ErrorStrategy.IN_STREAM to embed failures in the stream, or an error-handler callback to dispatch them separately. See Error handling for the full mode comparison.

Errors in the stream (IN_STREAM)

Use executeAsync(ErrorStrategy.IN_STREAM) (Java) or execute(on_error=ErrorStrategy.IN_STREAM) (Python) so single-key and batch async operations return failures as RecordResult entries instead of throwing:

Java

DataSet users = DataSet.of("test", "users");

RecordStream stream = session
    .update(users.id("exists")).bin("count").add(1)
    .update(users.id("no-such-key")).bin("count").add(1)
    .executeAsync(ErrorStrategy.IN_STREAM);

stream.forEach(result -> {
    if (result.isOk()) {
        System.out.println(result.key().userKey + ": success");
    } else {
        System.out.println(result.key().userKey + ": " + result.message());
    }
});

📖 API reference: DataSet.of(...) | DataSet.id(...) | Session.update(DataSet) | ChainableQueryBuilder.bin(...) | BinBuilder.add(...) | ChainableQueryBuilder.executeAsync(...) | RecordStream.forEach(...) | RecordResult.isOk() | ErrorStrategy.IN_STREAM

Python

users = DataSet.of("test", "users")

stream = await (
    session.update(users.id("exists")).bin("count").add(1)
    .update(users.id("no-such-key")).bin("count").add(1)
    .execute(on_error=ErrorStrategy.IN_STREAM)
)
try:
    async for result in stream:
        if result.is_ok:
            print(f"{result.key.value}: success")
        else:
            print(f"{result.key.value}: {result.exception or result.result_code}")
finally:
    stream.close()

📖 API reference: DataSet.of() | DataSet.id() | Session.update() | QueryBuilder.bin() | WriteSegmentBuilder.bin() | WriteSegmentBuilder.execute() | ErrorStrategy.IN_STREAM | RecordStream.close()

Error callback

Pass an error handler to executeAsync(handler) (Java) or execute(on_error=<callable>) (Python). Failures are dispatched to the callback and excluded from the returned stream:

Java

DataSet users = DataSet.of("test", "users");

RecordStream stream = session
    .update(users.id("exists")).bin("count").add(1)
    .update(users.id("no-such-key")).bin("count").add(1)
    .executeAsync((key, index, exception) ->
        System.err.println("Failed key " + key.userKey +
            " at index " + index + ": " + exception.getMessage())
    );

// Stream contains only successful results
stream.forEach(result ->
    System.out.println(result.key().userKey + ": OK")
);

📖 API reference: DataSet.of(...) | DataSet.id(...) | ChainableQueryBuilder.bin(...) | BinBuilder.add(...) | ChainableQueryBuilder.executeAsync(...) | RecordStream.forEach(...)

Python

users = DataSet.of("test", "users")

def handle_error(key, index, ex):
    print(f"Failed key {key.value} at index {index}: {ex}")

stream = await (
    session.update(users.id("exists")).bin("count").add(1)
    .update(users.id("no-such-key")).bin("count").add(1)
    .execute(on_error=handle_error)
)
try:
    async for result in stream:
        print(f"{result.key.value}: OK")
finally:
    stream.close()

📖 API reference: DataSet.of() | DataSet.id() | Session.update() | QueryBuilder.bin() | WriteSegmentBuilder.bin() | WriteSegmentBuilder.execute() | RecordStream.close()

API reference summary

Method	Description
Java: .executeAsync(ErrorStrategy.IN_STREAM)	Non-blocking execution; failures embedded in the stream
Java: .executeAsync(ErrorHandler)	Non-blocking execution; failures dispatched to callback
Python: await … .execute(on_error=ErrorStrategy.IN_STREAM)	Non-blocking execution; failures embedded in the stream
Python: await … .execute(on_error=<callable>)	Non-blocking execution; failures dispatched to callback
Python: await <builder>.execute() on an aerospike_sdk.aio session/builder	Non-blocking execution that returns a RecordStream
CompletableFuture + virtual threads	Compose parallel synchronous execute() calls (Java)
CompletableFuture.allOf()	Wait for multiple futures (Java)
await	Wait for async result (Python)
asyncio.gather()	Wait for multiple tasks (Python)
AsyncPool	Multi-loop async on free-threaded Python
session.get() / session.put()	Fast-path single-key API (Python, sync and async)

Next steps

Batch Operations

Efficient multi-record operations.

Batch Operations →

Behaviors

Configure timeouts and retries for async.

Behaviors →

Error Handling

Handle async errors gracefully.

Error Handling →