“

Series: S8 Technology Deep Dives · Article 14 | Level: Advanced | Reading Time: 20 min

Quarkus Reactive Deep Dive: Data Layer's Reactive Architecture

#TL;DR

Quarkus 3.x is the core framework for coomia-dip's Data Layer (Data Layer), leveraging reactive programming to achieve high-throughput, low-latency data services
This article deeply analyzes Quarkus's Vert.x event loop model, Mutiny reactive API, RESTEasy Reactive, Hibernate Reactive with Panache, and GraalVM Native Image compilation
Covers Quarkus + gRPC integration, reactive database access, backpressure handling, and coomia-dip performance benchmarks

#1. Quarkus in coomia-dip Data Layer

#1.1 Why Quarkus?

Dimension	Spring Boot	Quarkus
Startup time	3-10s	0.5-2s (JVM) / 0.02s (Native)
Memory footprint	200-500 MB	50-150 MB (JVM) / 20-50 MB (Native)
Reactive support	WebFlux (optional)	Native (Vert.x core)
Compile-time optimization	Limited	Extensive (ArC CDI, build-time augmentation)
GraalVM compatibility	Requires heavy config	First-class citizen
Use case	Control Layer (Control Layer)	Data Layer (Data Layer)

#1.2 Data Layer Service Matrix

Code

Data Layer Services (Quarkus 3.x):
│
├── QueryService          — OQL query execution
├── SearchService         — Full-text search
├── AnalyticsQueryService — Aggregation analytics
├── SubscriptionService   — Real-time subscription push
├── PipelineService       — Data pipeline management
├── MaterializationService — Materialized views
└── StorageService        — Iceberg/Doris storage abstraction

#2. Vert.x Event Loop Model

#2.1 Event Loop Architecture

Code

┌─────────────────────────────────────────────┐
│              Quarkus Application            │
│                                              │
│  ┌──────────────────────────────────────┐   │
│  │          IO Thread Pool              │   │
│  │  (Event Loop Threads = CPU cores)    │   │
│  │                                      │   │
│  │  Thread-0 ─── Event Loop ──→ Handle  │   │
│  │  Thread-1 ─── Event Loop ──→ Handle  │   │
│  │  Thread-N ─── Event Loop ──→ Handle  │   │
│  └──────────────────────────────────────┘   │
│                                              │
│  ┌──────────────────────────────────────┐   │
│  │         Worker Thread Pool           │   │
│  │   (for blocking operations)          │   │
│  │                                      │   │
│  │  Worker-0 ─── Blocking Task          │   │
│  │  Worker-1 ─── Blocking Task          │   │
│  └──────────────────────────────────────┘   │
└─────────────────────────────────────────────┘

#2.2 The Golden Rule of Event Loops

Java

// ❌ NEVER execute blocking operations on event loop threads
@Path("/query")
public class QueryResource {

    @GET
    public Uni<QueryResult> query(@QueryParam("oql") String oql) {
        // ❌ Blocking call freezes the event loop
        // return Uni.createFrom().item(blockingDbQuery(oql));

        // ✅ Use reactive client
        return reactiveClient.execute(oql)
            .map(rows -> QueryResult.from(rows));
    }
}

// ✅ If blocking is necessary, annotate with @Blocking
@Path("/legacy")
public class LegacyResource {

    @GET
    @Blocking  // Automatically dispatched to Worker thread pool
    public QueryResult legacyQuery(@QueryParam("sql") String sql) {
        return jdbcClient.query(sql);  // Blocking JDBC call
    }
}

#2.3 Thread Model Configuration

PROPERTIES

# application.properties

# IO thread count (default = CPU cores)
quarkus.vertx.event-loops-pool-size=8

# Worker thread pool size
quarkus.vertx.worker-pool-size=20

# Max event loop execution time (timeout warning)
quarkus.vertx.max-event-loop-execute-time=2s
quarkus.vertx.warning-exception-time=2s

#3. Mutiny Reactive API

#3.1 Uni and Multi

Java

// Uni<T> — Represents an async operation yielding 0 or 1 elements
Uni<OntologyInstance> uni = instanceRepository
    .findById(instanceId);

// Multi<T> — Represents an async stream of 0 to N elements
Multi<OntologyInstance> multi = instanceRepository
    .findByType(objectType);

#3.2 Operator Composition

Java

@ApplicationScoped
public class OntologyQueryService {

    @Inject
    ReactiveOntologyRepository repository;

    @Inject
    ReactiveCacheService cache;

    public Uni<QueryResult> executeQuery(OqlQuery query) {
        return parseQuery(query)
            .chain(parsed -> {
                // Check cache first
                return cache.get(parsed.cacheKey())
                    .onItem().ifNull().switchTo(() ->
                        // Cache miss, query database
                        repository.execute(parsed)
                            .chain(result ->
                                // Write result to cache
                                cache.put(parsed.cacheKey(), result)
                                    .replaceWith(result)
                            )
                    );
            })
            .onFailure().retry()
                .withBackOff(Duration.ofMillis(100), Duration.ofSeconds(1))
                .atMost(3)
            .onFailure().recoverWithItem(error -> {
                log.error("Query failed", error);
                return QueryResult.error(error.getMessage());
            });
    }

    public Multi<OntologyInstance> streamInstances(String objectType) {
        return repository.findByType(objectType)
            .select().where(instance -> instance.isActive())
            .onItem().transform(instance -> enrichWithDerivedProperties(instance))
            .group().intoLists().of(100)  // Batch every 100 items
            .onItem().transformToUniAndMerge(batch ->
                processBatch(batch)
            );
    }
}

#3.3 Concurrent Operations

Java

// Execute multiple async operations in parallel
public Uni<EnrichedInstance> enrichInstance(String instanceId) {
    Uni<OntologyInstance> instanceUni = repository.findById(instanceId);
    Uni<List<LinkType>> linksUni = linkRepository.findBySourceId(instanceId);
    Uni<Map<String, Object>> metricsUni = metricService.getMetrics(instanceId);

    return Uni.combine().all()
        .unis(instanceUni, linksUni, metricsUni)
        .with((instance, links, metrics) ->
            new EnrichedInstance(instance, links, metrics)
        );
}

#4. RESTEasy Reactive

#4.1 External REST API

Java

@Path("/api/v1/ontology")
@Produces(MediaType.APPLICATION_JSON)
@ApplicationScoped
public class OntologyResource {

    @Inject
    OntologyQueryService queryService;

    @GET
    @Path("/instances/{objectType}")
    public Multi<OntologyInstance> listInstances(
            @PathParam("objectType") String objectType,
            @QueryParam("limit") @DefaultValue("100") int limit,
            @QueryParam("offset") @DefaultValue("0") int offset) {
        return queryService.findByType(objectType)
            .skip().first(offset)
            .select().first(limit);
    }

    @POST
    @Path("/query")
    @Consumes(MediaType.APPLICATION_JSON)
    public Uni<QueryResult> executeQuery(OqlQueryRequest request) {
        return queryService.executeQuery(request.toOqlQuery());
    }

    @GET
    @Path("/instances/{objectType}/{instanceId}")
    @RestStreamElementType(MediaType.APPLICATION_JSON)
    public Multi<ServerSentEvent<OntologyEvent>> streamEvents(
            @PathParam("objectType") String objectType,
            @PathParam("instanceId") String instanceId) {
        // Server-Sent Events stream
        return subscriptionService.subscribe(objectType, instanceId)
            .map(event -> Sse.event(event).id(event.getId()));
    }
}

#4.2 Request Filters

Java

@Provider
@Priority(Priorities.AUTHENTICATION)
public class WorldContextFilter implements ContainerRequestFilter {

    @Override
    public void filter(ContainerRequestContext requestContext) {
        String worldId = requestContext.getHeaderString("X-World-Id");
        if (worldId == null || worldId.isBlank()) {
            requestContext.abortWith(
                Response.status(400)
                    .entity(Map.of("error", "X-World-Id header required"))
                    .build()
            );
            return;
        }
        WorldContext.setCurrent(worldId);
    }
}

#5. Hibernate Reactive and Panache

#5.1 Reactive Repository

Java

@ApplicationScoped
public class OntologyInstanceRepository
        implements PanacheRepositoryBase<OntologyInstanceEntity, String> {

    public Uni<OntologyInstanceEntity> findByObjectId(String objectId) {
        return find("objectId", objectId).firstResult();
    }

    public Multi<OntologyInstanceEntity> findByType(String objectType) {
        return find("objectType", objectType).stream();
    }

    public Uni<Long> countByType(String objectType) {
        return count("objectType", objectType);
    }

    public Uni<List<OntologyInstanceEntity>> search(
            String objectType,
            Map<String, Object> filters,
            int limit,
            int offset) {
        StringBuilder query = new StringBuilder("objectType = :type");
        Parameters params = Parameters.with("type", objectType);

        for (Map.Entry<String, Object> filter : filters.entrySet()) {
            query.append(" AND properties->>'")
                 .append(filter.getKey())
                 .append("' = :")
                 .append(filter.getKey());
            params.and(filter.getKey(), filter.getValue().toString());
        }

        return find(query.toString(), params)
            .page(Page.of(offset / limit, limit))
            .list();
    }
}

#5.2 Entity Definition

Java

@Entity
@Table(name = "ontology_instances")
public class OntologyInstanceEntity extends PanacheEntityBase {

    @Id
    public String id;

    @Column(name = "object_type", nullable = false)
    public String objectType;

    @Column(name = "object_id", nullable = false, unique = true)
    public String objectId;

    @Column(name = "world_id", nullable = false)
    public String worldId;

    @Type(JsonBinaryType.class)
    @Column(name = "properties", columnDefinition = "jsonb")
    public Map<String, Object> properties;

    @Column(name = "version")
    @Version
    public long version;

    @CreationTimestamp
    @Column(name = "created_at")
    public Instant createdAt;

    @UpdateTimestamp
    @Column(name = "updated_at")
    public Instant updatedAt;
}

#6. gRPC Service Integration

#6.1 gRPC Service Definition

Java

@GrpcService
public class QueryGrpcService extends MutinyQueryServiceGrpc.QueryServiceImplBase {

    @Inject
    OntologyQueryService queryService;

    @Override
    public Uni<QueryResponse> executeQuery(QueryRequest request) {
        return queryService.executeQuery(toOqlQuery(request))
            .map(result -> QueryResponse.newBuilder()
                .addAllInstances(result.getInstances().stream()
                    .map(this::toProto)
                    .toList())
                .setTotalCount(result.getTotalCount())
                .build()
            );
    }

    @Override
    public Multi<InstanceEvent> streamChanges(StreamRequest request) {
        return queryService.streamChanges(
            request.getObjectType(),
            request.getFilterExpression()
        ).map(this::toProtoEvent);
    }
}

#6.2 gRPC Client

Java

@ApplicationScoped
public class ControlPlaneClient {

    @GrpcClient("control-Layer")
    MutinyOntologyServiceGrpc.MutinyOntologyServiceStub ontologyStub;

    public Uni<OntologySchema> getSchema(String objectType) {
        return ontologyStub.getSchema(
            GetSchemaRequest.newBuilder()
                .setObjectType(objectType)
                .build()
        ).map(this::fromProto);
    }
}

PROPERTIES

# gRPC client configuration
quarkus.grpc.clients.control-Layer.host=control-Layer-service
quarkus.grpc.clients.control-Layer.port=9090
quarkus.grpc.clients.control-Layer.plain-text=true

#7. Reactive Messaging

#7.1 Kafka Reactive Integration

Java

@ApplicationScoped
public class CdcEventProcessor {

    @Incoming("cdc-events")
    @Outgoing("processed-events")
    public Multi<Record<String, ProcessedEvent>> process(
            Multi<Record<String, CdcEvent>> events) {
        return events
            .onItem().transformToUniAndMerge(record -> {
                CdcEvent event = record.value();
                return enrichEvent(event)
                    .map(enriched -> Record.of(
                        record.key(),
                        new ProcessedEvent(enriched)
                    ));
            })
            .select().where(record -> record.value().isValid());
    }
}

PROPERTIES

# Kafka connector configuration
mp.messaging.incoming.cdc-events.connector=smallrye-kafka
mp.messaging.incoming.cdc-events.topic=coomia-dip.cdc.events
mp.messaging.incoming.cdc-events.value.deserializer=io.quarkus.kafka.client.serialization.JsonbDeserializer
mp.messaging.incoming.cdc-events.group.id=data-Layer-processor
mp.messaging.incoming.cdc-events.auto.offset.reset=latest
mp.messaging.incoming.cdc-events.failure-strategy=dead-letter-queue

#8. Backpressure and Flow Control

#8.1 Mutiny Backpressure Handling

Java

public Multi<OntologyInstance> streamWithBackpressure(String objectType) {
    return repository.findByType(objectType)
        // Control downstream consumption rate
        .onOverflow()
            .buffer(1000)        // Buffer 1000 items
            .drop()              // Drop new elements when buffer full
        // Or use pacing strategy
        .paceDemand()
            .using(1, Duration.ofMillis(10));  // Request 1 item every 10ms
}

#8.2 gRPC Stream Backpressure

Java

@GrpcService
public class StreamingGrpcService
        extends MutinyStreamServiceGrpc.StreamServiceImplBase {

    @Override
    public Multi<DataChunk> streamData(StreamRequest request) {
        return dataService.streamData(request.getQuery())
            .onItem().transform(this::toChunk)
            // gRPC streams automatically support backpressure (HTTP/2 Flow Control)
            .onOverflow().buffer(256);
    }
}

#9. GraalVM Native Image

#9.1 Native Compilation Configuration

PROPERTIES

# application.properties
quarkus.native.enabled=true
quarkus.native.container-build=true
quarkus.native.builder-image=quay.io/quarkus/ubi-quarkus-mandrel-builder-image:jdk-21

quarkus.native.additional-build-args=\
  --initialize-at-run-time=io.netty.handler.ssl.BoringSSL,\
  -H:+ReportExceptionStackTraces

#9.2 Performance Comparison

Metric	JVM Mode	Native Mode
Startup time	1.5s	0.025s
First request latency	200 ms	5 ms
Memory footprint (RSS)	180 MB	35 MB
Peak throughput	15,000 RPS	12,000 RPS
Build time	5s	3 min

#9.3 Native Image Limitations and Workarounds

Limitation	Workaround
Reflection restricted	Use `@RegisterForReflection`
Dynamic proxy restricted	Quarkus ArC compile-time CDI
JNI restricted	Avoid JNI libraries
Serialization restricted	Use Quarkus serialization extensions

#10. Performance Benchmarks and Tuning

#10.1 Benchmark Results

Code

Test environment: 4 cores 8GB, PostgreSQL + Redis
Tool: wrk2, 60-second duration

Simple query (single instance GET):
  Quarkus Reactive: 28,000 RPS, P99 = 3.2ms
  Spring Boot MVC:   8,500 RPS, P99 = 12ms

Complex query (OQL + Join):
  Quarkus Reactive: 4,500 RPS, P99 = 45ms
  Spring Boot MVC:  1,800 RPS, P99 = 120ms

gRPC streaming (10,000 records):
  Quarkus Reactive: 2.1s to complete
  Traditional REST pagination: 8.5s to complete

#10.2 Key Tuning Parameters

PROPERTIES

# Connection pool
quarkus.datasource.reactive.max-size=20
quarkus.datasource.reactive.idle-removal-interval=5m

# HTTP server
quarkus.http.io-threads=8
quarkus.http.limits.max-body-size=10M
quarkus.http.idle-timeout=30s

# gRPC
quarkus.grpc.server.max-inbound-message-size=10485760
quarkus.grpc.server.handshake-timeout=10s

# Redis
quarkus.redis.max-pool-size=32
quarkus.redis.max-pool-waiting=64

#11. Key Takeaways

Topic	Key Conclusion
Framework choice	Data Layer uses Quarkus, Control Layer uses Spring Boot
Event loop	Never block on IO threads; use @Blocking when necessary
Mutiny	Uni for single-value async, Multi for stream async
Database	Hibernate Reactive + Panache provides reactive ORM
gRPC	Quarkus natively supports gRPC Server + Client
Messaging	SmallRye Reactive Messaging integrates Kafka
Native	60x faster startup, 80% less memory, but slower builds
Performance	3-4x throughput improvement over Spring Boot

“

Next up: S8-15 dives into FastAPI + gRPC dual-protocol services, exploring how coomia-dip Intelligence Layer exposes both REST and gRPC interfaces simultaneously.