🍳 Virtual Threads Cookbook

Java 21+ Virtual Threads with Enterprise Technologies

A comprehensive guide for using virtual threads with MongoDB, Cassandra, Kafka,
Elasticsearch, Protocol Buffers, and CQRS frameworks.

📋 Table of Contents
Overview
MongoDB with Virtual Threads
Cassandra with Virtual Threads
Kafka with Virtual Threads
Elasticsearch with Virtual Threads
Protocol Buffers with Virtual Threads
CQRS with Axon and Virtual Threads
Comprehensive Demo
Key Benefits
Best Practices

### Technologies Covered

NoSQL DBs: MongoDB, Cassandra, HBase, Zookeeper

Serialization: Thrift, Protocol Buffers
Data Processing: Hadoop, Kafka
Frameworks: Guice, Spring, Axon (CQRS)
Search: Lucene, Elasticsearch

MongoDB with Virtual Threads

Bulk Message Archiving

 @Service
public class MongoVirtualThreadService {
private final MongoTemplate mongoTemplate;
private final ExecutorService virtualExecutor;

@Inject
public MongoVirtualThreadService(MongoTemplate mongoTemplate) {
[Link] = mongoTemplate;
[Link] = [Link]();
}

// Bulk message archiving with virtual threads

public CompletableFuture<List<String>> bulkArchiveMessages(List<ArchivedMessage>
messages) {
[Link]("🗄️ Archiving " + [Link]() + " messages with
virtual threads");

List<CompletableFuture<String>> futures = [Link]()

.map(message -> [Link](() -> {
try {
// Each message gets its own virtual thread for DB operation
[Link](message);

// Simulate compliance checking (I/O operation)

[Link]([Link](50));

return "Archived: " + [Link]() + " on " +

[Link]();
} catch (Exception e) {
return "Failed: " + [Link]() + " - " + [Link]();
}
}, virtualExecutor))
.toList();

return [Link]([Link](new CompletableFuture[0]))

.thenApply(v -> [Link]()
.map(CompletableFuture::join)
.toList());
}
}

Parallel Aggregations
 // Parallel aggregation across multiple collections
public CompletableFuture<Map<String, Object>> parallelAggregations(String userId) {
// Each aggregation runs in its own virtual thread
CompletableFuture<Long> messageCount = [Link](() -> {
Query query = new Query([Link]("userId").is(userId));
return [Link](query, [Link]);
}, virtualExecutor);

CompletableFuture<List<TransactionSummary>> transactionSummary =
[Link](() -> {
Aggregation agg = [Link](
[Link]([Link]("userId").is(userId)),
[Link]("currency").sum("amount").as("total")
);
return [Link](agg, "transactions", [Link])
.getMappedResults();
}, virtualExecutor);

// Combine all results

return [Link](messageCount, transactionSummary)
.thenApply(v -> [Link](
"messageCount", [Link](),
"transactionSummary", [Link]()
));
}

Cassandra with Virtual Threads

High-Throughput Message Ingestion

@Service
public class CassandraVirtualThreadService {
private final CqlSession session;
private final PreparedStatement insertStatement;

// High-throughput message ingestion

public CompletableFuture<String> massiveMessageIngestion(List<UserMessage>
messages) {
📡
[Link](" Ingesting " + [Link]() + " messages to
Cassandra");

try (ExecutorService executor = [Link]())

{

List<CompletableFuture<Void>> insertFutures = [Link]()

.map(message -> [Link](() -> {
try {
BoundStatement bound = [Link](
[Link](),
[Link](),
[Link](),
[Link]()
);

// Cassandra async operation in virtual thread

[Link](bound)
.toCompletableFuture()
.join(); // Block this virtual thread until complete

} catch (Exception e) {
throw new RuntimeException("Failed to insert message: " +
[Link](), e);
}
}, executor))
.toList();

return [Link]([Link](new
CompletableFuture[0]))
.thenApply(v -> "Successfully ingested " + [Link]() + "
messages");
}
}
}
Parallel Time Range Queries

// Parallel data retrieval across time ranges

public CompletableFuture<List<UserMessage>> parallelTimeRangeQuery(
String userId, List<TimeRange> timeRanges) {

try (ExecutorService executor = [Link]()) {

List<CompletableFuture<List<UserMessage>>> rangeFutures = [Link]()

.map(range -> [Link](() -> {
BoundStatement bound = [Link](
userId, [Link](), [Link]()
);

return [Link](bound)
.toCompletableFuture()
.join()
.map(row -> mapRowToUserMessage(row))
.all();
}, executor))
.toList();

return [Link]([Link](new
CompletableFuture[0]))
.thenApply(v -> [Link]()
.flatMap(future -> [Link]().stream())
.toList());
}
}

Kafka with Virtual Threads

Massive Parallel Publishing

@Service
public class KafkaVirtualThreadService {
private final KafkaTemplate<String, byte[]> kafkaTemplate;
private final MessageSerializationService serializationService;

// Massive parallel message publishing

public CompletableFuture<String>
publishFinancialEvents(List<FinancialTransaction> transactions) {
[Link]("📤 Publishing " + [Link]() + " events to
Kafka");

try (ExecutorService executor = [Link]())

{

List<CompletableFuture<SendResult<String, byte[]>>> publishFutures =

[Link]()
.map(transaction -> [Link](() -> {
try {
// Serialize in virtual thread
byte[] serializedData =
[Link](transaction);

ProducerRecord<String, byte[]> record = new ProducerRecord<>(

"financial-transactions",
[Link](),
serializedData
);

// Kafka send is async, but we wait in this virtual thread

return [Link](record).get();

} catch (Exception e) {
throw new RuntimeException("Failed to publish: " +
[Link](), e);
}
}, executor))
.toList();

return [Link]([Link](new
CompletableFuture[0]))
.thenApply(v -> {
long successCount = [Link]()
.mapToLong(future -> [Link]() ? 0 :
1)
.sum();
 return "Published " + successCount + "/" + [Link]() +
" events";
});
}
}
}

Fan-out Processing

// Fan-out processing to multiple topics

public CompletableFuture<Void> fanOutProcessing(FinancialTransaction transaction) {
try (ExecutorService executor = [Link]()) {

// Send to different topics based on business rules

List<CompletableFuture<Void>> fanOutTasks = [Link](

// Compliance topic
[Link](() -> {
if ([Link]() > 0.7) {
publishToTopic("compliance-alerts", transaction);
}
}, executor),

// Reporting topic
[Link](() -> {
publishToTopic("financial-reporting", transaction);
}, executor),

// High-value transactions
[Link](() -> {
if ([Link]().compareTo(new BigDecimal("10000")) > 0) {
publishToTopic("high-value-transactions", transaction);
}
}, executor)
);

return [Link]([Link](new
CompletableFuture[0]));
}
}

Elasticsearch with Virtual Threads

Bulk Document Indexing
@Service
public class ElasticsearchVirtualThreadService {
private final ElasticsearchRestTemplate elasticsearchTemplate;

// Parallel indexing for massive document sets

public CompletableFuture<IndexingResult>
bulkIndexDocuments(List<FinancialMessageDocument> documents) {
[Link](" 🔍 Indexing " + [Link]() + " documents to
Elasticsearch");

try (ExecutorService executor = [Link]())

{

// Batch documents for efficient indexing

int batchSize = 100;
List<List<FinancialMessageDocument>> batches = partitionList(documents,
batchSize);

List<CompletableFuture<Integer>> batchFutures = [Link]()

.map(batch -> [Link](() -> {
try {
// Each batch gets its own virtual thread
[Link](batch);
return [Link]();
} catch (Exception e) {
[Link]("Batch indexing failed: " +
[Link]());
return 0;
}
}, executor))
.toList();

return [Link]([Link](new
CompletableFuture[0]))
.thenApply(v -> {
int totalIndexed = [Link]()
.mapToInt(CompletableFuture::join)
.sum();
return new IndexingResult(totalIndexed, [Link]());
});
}
}
}
Parallel Multi-Index Search

// Parallel search across multiple indices

public CompletableFuture<Map<String, SearchResults>> parallelMultiIndexSearch(
String searchTerm, List<String> indices) {

try (ExecutorService executor = [Link]()) {

Map<String, CompletableFuture<SearchResults>> searchFutures =

[Link]()
.collect([Link](
index -> index,
index -> [Link](() -> {
try {
// Each index search in its own virtual thread
Query query = [Link]()
.withQuery([Link](searchTerm,
"content", "subject"))
.build();

SearchHits<FinancialMessageDocument> hits =
[Link](query,
[Link]);

return new SearchResults([Link](),

[Link]());

} catch (Exception e) {
return new SearchResults(0, [Link]());
}
}, executor)
));

return [Link]([Link]().toArray(new
CompletableFuture[0]))
.thenApply(v -> [Link]().stream()
.collect([Link](
[Link]::getKey,
entry -> [Link]().join()
)));
}
}

Protocol Buffers with Virtual Threads

Parallel Serialization

@Service
public class ProtocolBufferVirtualThreadService {

// Parallel serialization of large datasets

public CompletableFuture<List<byte[]>>
parallelSerialization(List<FinancialTransaction> transactions) {
[Link]("🔄 Serializing " + [Link]() + " transactions
with Protocol Buffers");

try (ExecutorService executor = [Link]())

{

List<CompletableFuture<byte[]>> serializationFutures =
[Link]()
.map(transaction -> [Link](() -> {
try {
// Each serialization in its own virtual thread
return [Link]()
.setTransactionId([Link]())
.setUserId([Link]())
.setAmount([Link]().toString())
.setCurrency([Link]())
.setTimestamp([Link]().toEpochMilli())
.build()
.toByteArray();
} catch (Exception e) {
throw new RuntimeException("Serialization failed for: " +
[Link](), e);
}
}, executor))
.toList();

return [Link]([Link](new
CompletableFuture[0]))
.thenApply(v -> [Link]()
.map(CompletableFuture::join)
.toList());
}
}
}
Parallel Deserialization and Validation

// Parallel deserialization and validation

public CompletableFuture<List<FinancialTransaction>>
parallelDeserialization(List<byte[]> serializedData) {
try (ExecutorService executor = [Link]()) {

List<CompletableFuture<FinancialTransaction>> deserializationFutures =
[Link]()
.map(data -> [Link](() -> {
try {
TransactionProto proto = [Link](data);

// Validation in virtual thread

validateProtoMessage(proto);

return [Link]()
.transactionId([Link]())
.userId([Link]())
.amount(new BigDecimal([Link]()))
.currency([Link]())
.timestamp([Link]([Link]()))
.build();
} catch (Exception e) {
throw new RuntimeException("Deserialization failed", e);
}
}, executor))
.toList();

return [Link]([Link](new
CompletableFuture[0]))
.thenApply(v -> [Link]()
.map(CompletableFuture::join)
.toList());
}
}

CQRS with Axon and Virtual Threads

Parallel Command Processing

 @Service
public class CQRSVirtualThreadService {
private final CommandGateway commandGateway;
private final QueryGateway queryGateway;

// Parallel command processing

public CompletableFuture<List<String>>
processTransactionBatch(List<CreateTransactionCommand> commands) {
[Link]("⚡ Processing " + [Link]() + " commands with
CQRS");

try (ExecutorService executor = [Link]())

{

List<CompletableFuture<String>> commandFutures = [Link]()

.map(command -> [Link](() -> {
try {
// Each command in its own virtual thread
return [Link](command);
} catch (Exception e) {
return "Failed: " + [Link]() + " - " +
[Link]();
}
}, executor))
.toList();

return [Link]([Link](new
CompletableFuture[0]))
.thenApply(v -> [Link]()
.map(CompletableFuture::join)
.toList());
}
}
}

Parallel Query Execution

 // Parallel query execution
public CompletableFuture<Map<String, Object>> parallelReportGeneration(String userId)
{
try (ExecutorService executor = [Link]()) {

// Multiple queries in parallel virtual threads

CompletableFuture<List<TransactionView>> transactions =
[Link](() -> {
FindTransactionsByUserQuery query = new
FindTransactionsByUserQuery(userId);
return [Link](query,
[Link]([Link])).join();
}, executor);

CompletableFuture<UserProfileView> userProfile =
[Link](() -> {
FindUserProfileQuery query = new FindUserProfileQuery(userId);
return [Link](query,
[Link]([Link])).join();
}, executor);

CompletableFuture<List<ComplianceAlert>> alerts =
[Link](() -> {
FindComplianceAlertsQuery query = new FindComplianceAlertsQuery(userId);
return [Link](query,
[Link]([Link])).join();
}, executor);

return [Link](transactions, userProfile, alerts)

.thenApply(v -> [Link](
"transactions", [Link](),
"userProfile", [Link](),
"complianceAlerts", [Link]()
));
}
}

Comprehensive Demo
@Component
public class VirtualThreadCookbookDemo {

public void runComprehensiveDemo() {

🍳
[Link](" Virtual Threads Cookbook - Tech Stack Demo");

[Link]("=========================================================");

// Simulate massive concurrent processing

try (ExecutorService executor = [Link]())
{

Instant start = [Link]();

// Create sample data

List<FinancialTransaction> transactions =
createSampleTransactions(10_000);
List<ArchivedMessage> messages = createSampleMessages(5_000);

// Parallel processing across all technologies

CompletableFuture<String> mongoArchiving =
[Link](() -> {
[Link]("📁 MongoDB: Archiving messages...");
return "MongoDB: Archived " + [Link]() + " messages";
}, executor);

CompletableFuture<String> cassandraIngestion =
[Link](() -> {
[Link]("🗃️ Cassandra: Ingesting time-series data...");
return "Cassandra: Ingested " + [Link]() + "
transactions";
}, executor);

CompletableFuture<String> kafkaPublishing =
[Link](() -> {
[Link]("📡 Kafka: Publishing events...");
return "Kafka: Published " + [Link]() + " events";
}, executor);

CompletableFuture<String> elasticsearchIndexing =
[Link](() -> {
[Link]("🔍 Elasticsearch: Indexing documents...");
return "Elasticsearch: Indexed " + [Link]() + " documents";
}, executor);
 // Wait for all operations to complete
[Link](
mongoArchiving, cassandraIngestion, kafkaPublishing,
elasticsearchIndexing
).join();

Duration elapsed = [Link](start, [Link]());

[Link]("\n ✅ All operations completed in " +

[Link]() + "ms");
[Link](" 🚀Virtual threads enabled massive parallel
processing!");
}
}
}

Key Benefits

🚀 Massive Scalability
Handle 10,000+ concurrent operations with minimal memory footprint
Perfect for I/O-heavy workloads common in financial services
No thread pool exhaustion with database connections

⚡ Real-world Performance Improvements

Message Archiving: Process 100,000 emails/chats simultaneously
Compliance Scanning: Parallel analysis across millions of documents
Risk Assessment: Concurrent processing of financial transactions
eDiscovery: Search across terabytes of data in parallel

🎯 Best Practices
✅ Do's
Use try-with-resources for ExecutorService auto-cleanup
Leverage virtual threads for I/O-intensive operations
Combine with async APIs (Cassandra, MongoDB async drivers)
Use for high-concurrency scenarios (10,000+ operations)
Structure code with CompletableFuture for composability

❌
❌ Don'ts
Avoid for CPU-intensive tasks (use ForkJoinPool instead)
Don't use excessive synchronized blocks (can pin virtual threads)
Avoid blocking on virtual threads unnecessarily
Don't create virtual threads for short-lived operations

💡 Performance Tips
Batch operations when possible (Elasticsearch bulk indexing)
Use appropriate timeouts for I/O operations
Monitor virtual thread metrics in production
Consider using virtual thread factories for custom naming

Conclusion

Virtual threads transform enterprise Java applications from being limited by thread pool sizes
to being limited only by business logic and I/O capacity. For 's technology stack, this means:

Unprecedented scalability for message archiving and compliance systems

Simplified concurrency models without complex thread pool management
Better resource utilization across MongoDB, Cassandra, Kafka, and Elasticsearch
Enhanced system responsiveness under high load conditions

This cookbook demonstrates how virtual threads enable massive parallel processing that would
be impossible with traditional platform threads, making them perfect for the high-throughput, I/O-
intensive workloads typical in financial services technology.