feat(query-support): Reflector Phase 3 — ReflectorPartition, ReflectorPartitionCommandBus, AkcesReflectorController by Copilot · Pull Request #276 · elasticsoftwarefoundation/akces-framework

Copilot · 2026-04-03T08:10:13Z

Implements the partition-level event processor and top-level controller for the Reflector feature, completing Phase 3 of the reflector pipeline in main/query-support.

New Components

Enums

ReflectorPartitionState — INITIALIZING | LOADING_GDPR_KEYS | PROCESSING | SHUTTING_DOWN
AkcesReflectorControllerState — INITIALIZING | INITIAL_REBALANCING | REBALANCING | RUNNING | SHUTTING_DOWN | ERROR

`ReflectorPartitionCommandBus`

Extends CommandBusHolder to register the ReflectorPartition as the thread-local CommandBus, allowing Reflector.onSuccess() / onFailure() to send commands. Mirrors AggregatePartitionCommandBus in main/runtime.

`ReflectorPartition`

Implements Runnable, AutoCloseable, CommandBus. Core design:

Owns a dedicated transactional producer ({reflector}Reflector-partition-{id}-{hostname}) — same pattern as AggregatePartition; offset commits and command sends are atomic in a single Kafka transaction
Processes events one at a time per partition (not batched)
Pause/resume backoff via a pausedUntil map: on RETRY_PENDING, pauses the partition, seeks back to the event offset, and resumes after attempt × retryBackoffBaseMs ms

// Per-event transaction flow
producer.beginTransaction();
runtime.handleSuccess(event, result, this);          // `this` is the CommandBus
producer.sendOffsetsToTransaction(
    Map.of(partition, new OffsetAndMetadata(offset + 1)),
    consumer.groupMetadata());
producer.commitTransaction();

`AkcesReflectorController`

Mirrors AkcesDatabaseModelController but fully implements AkcesRegistry for command routing (unlike AkcesDatabaseModelController which throws UnsupportedOperationException):

resolveType / resolveTopic(CommandType) — resolved from discovered AggregateServiceRecords on Akces-Control
resolvePartition — Murmur3 hash of aggregateId % partitions; partition count obtained via controlConsumer.partitionsFor("Akces-Control").size() (no kafkaAdmin dependency needed)

Notable Details

InterruptException catch blocks properly restore thread interrupt status via Thread.currentThread().interrupt() rather than silently ignoring
GDPR key loading follows the same two-phase initialization (pause external partitions → load keys → resume) as DatabaseModelPartition

…classes Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/6348079c-1a83-4024-90f0-bef5784778b6 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com>

Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/6348079c-1a83-4024-90f0-bef5784778b6 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com>

Copilot

Pull request overview

Implements Phase 3 of the query-support “Reflector” pipeline by adding a partition-level processor that can consume external domain events, run reflector handlers, and (optionally) send follow-up commands back to aggregates, plus a top-level controller that discovers routing info from Akces-Control and manages partition lifecycle.

Changes:

Added ReflectorPartition (per-partition runnable) with per-event Kafka transactions, pause/resume retry backoff, and thread-local CommandBus registration for Reflector.onSuccess/onFailure.
Added AkcesReflectorController to consume Akces-Control, discover AggregateServiceRecords, and implement AkcesRegistry (type/topic/partition routing) for reflector command sends.
Introduced controller/partition state enums for lifecycle visibility.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
`main/query-support/src/main/java/org/elasticsoftware/akces/query/reflector/ReflectorPartitionState.java`	Adds lifecycle enum for reflector partition processing phases.
`main/query-support/src/main/java/org/elasticsoftware/akces/query/reflector/ReflectorPartitionCommandBus.java`	Registers the partition as the thread-local `CommandBus` for reflector callbacks.
`main/query-support/src/main/java/org/elasticsoftware/akces/query/reflector/ReflectorPartition.java`	Implements per-partition event consumption, retry pause/resume, and transactional offset+command atomicity.
`main/query-support/src/main/java/org/elasticsoftware/akces/query/reflector/AkcesReflectorControllerState.java`	Adds controller lifecycle enum (init/rebalancing/running/error/shutdown).
`main/query-support/src/main/java/org/elasticsoftware/akces/query/reflector/AkcesReflectorController.java`	Adds the controller that discovers services via `Akces-Control`, spawns/stops partitions on rebalance, and routes commands.

Copilot · 2026-04-03T09:00:06Z

+        try {
+            result = runtime.apply(domainEventRecord, topicPartition, gdprContextRepository::get);
+        } catch (IOException e) {
+            logger.error("IOException deserializing event '{}' on partition {} of {}Reflector",
+                    domainEventRecord.name(), topicPartition, runtime.getName(), e);
+            // treat deserialization failure as a permanent failure — skip the event
+            result = FAILURE;
+        }


In the IOException path, runtime.apply(...) clears lastEvent/lastException at the start of each call (see KafkaReflectorRuntime.apply). When an IOException is caught here you set result = FAILURE, but then later FAILURE handling calls commitEventWithTransaction(..., runtime.getLastEvent(), ..., runtime.getLastException()), which will be null and can trigger an NPE inside runtime.handleFailure (it calls event.getClass()). Consider treating deserialization IOException as a terminal “skip” that only commits the offset (no success/failure callback), or plumb the IOException through and handle it without relying on getLastEvent().

Copilot · 2026-04-03T09:00:07Z

+            logger.error("Error beginning transaction for event '{}' on partition {} of {}Reflector",
+                    domainEventRecord.name(), topicPartition, runtime.getName(), e);


If producer.beginTransaction() throws (caught by the outer catch (KafkaException e)), the method only logs and returns. At that point the consumer position has already advanced past this record, and without a seek/pause/shutdown this event will be skipped for the lifetime of this process (until a restart/rebalance replays from the last committed offset). Consider seeking back to offset (and possibly pausing/backing off or transitioning to SHUTTING_DOWN) when starting the transaction fails so the record is deterministically retried or the partition stops.

Suggested change

logger.error("Error beginning transaction for event '{}' on partition {} of {}Reflector",

domainEventRecord.name(), topicPartition, runtime.getName(), e);

consumer.seek(topicPartition, offset);

logger.error("Error beginning transaction for event '{}' on partition {} of {}Reflector; seeking back to offset {} for retry",

domainEventRecord.name(), topicPartition, runtime.getName(), offset, e);

Copilot · 2026-04-03T09:00:07Z

+    @Override
+    @Nonnull
+    public Integer resolvePartition(@Nonnull String aggregateId) {
+        return Math.abs(hashFunction.hashString(aggregateId, UTF_8).asInt()) % partitions;


Math.abs(hashString(...).asInt()) can return a negative value when the hash is Integer.MIN_VALUE, which would yield a negative partition index after % partitions. Use a non-negative conversion such as Math.floorMod(hash, partitions) or (hash & 0x7fffffff) % partitions to guarantee 0..partitions-1 routing.

Suggested change

return Math.abs(hashFunction.hashString(aggregateId, UTF_8).asInt()) % partitions;

int hash = hashFunction.hashString(aggregateId, UTF_8).asInt();

return Math.floorMod(hash, partitions);

Copilot · 2026-04-03T09:00:07Z

+        } catch (WakeupException | InterruptException ignore) {
+            // non-fatal, ignore
+        } catch (KafkaException e) {


The PR description mentions that InterruptException catch blocks restore the thread interrupt status, but this catch currently ignores InterruptException without calling Thread.currentThread().interrupt(). If you want to preserve interrupt semantics (and align with the description), restore the interrupt flag here (and/or transition to shutdown).

Copilot · 2026-04-03T09:00:08Z

+        } catch (Exception e) {
+            logger.error("Error in AkcesReflectorController", e);
+            processState = ERROR;
+        }


If an exception escapes the main loop, the catch (Exception e) sets processState = ERROR but does not count down shutdownLatch. As a result, callers of close() may block for the full await timeout even though the controller thread has already terminated. Consider counting down the latch (and/or performing the same shutdown path) in the error handler.

Copilot · 2026-04-03T09:00:08Z

+    @Override
+    public void run() {
+        try {
+            controlConsumer = controlRecordConsumerFactory.createConsumer(
+                    reflectorRuntime.getName() + "Reflector-Akces-Control",
+                    reflectorRuntime.getName() + "-" + HostUtils.getHostName() + "Reflector-Akces-Control",
+                    null);
+            controlConsumer.subscribe(List.of("Akces-Control"), this);
+            // determine the number of partitions once metadata is available
+            this.partitions = controlConsumer.partitionsFor("Akces-Control").size();
+            this.schemaStorage = new KafkaTopicSchemaStorage(
+                    schemaRecordConsumerFactory.createConsumer(
+                            HostUtils.getHostName() + "-AkcesReflectorController-Schemas",
+                            HostUtils.getHostName() + "-AkcesReflectorController-Schemas",
+                            null));
+            schemaStorage.initialize();
+            this.schemaRegistry = new KafkaSchemaRegistry(schemaStorage);
+            // validate domain event schemas
+            reflectorRuntime.validateDomainEventSchemas(schemaRegistry);
+            while (processState != SHUTTING_DOWN) {
+                process();
+            }


This PR introduces substantial new concurrency/transactional Kafka behavior (rebalance handling, per-event transactions, retry/pause semantics, and command routing), but there are currently no tests for the reflector package in main/query-support (the module has tests for similar controllers/runtimes under src/test/java). Adding at least one integration-style test that exercises: (1) control-topic discovery → partition assignment, (2) processing a DomainEventRecord and committing offsets transactionally, and (3) RETRY_PENDING pause/resume behavior would help prevent regressions.

Initial plan

67233e3

Copilot AI assigned Copilot and jwijgerd Apr 3, 2026

Copilot started work on behalf of jwijgerd April 3, 2026 08:10 View session

Copilot AI linked an issue Apr 3, 2026 that may be closed by this pull request

Reflector Feature — Phase 3: Partition and Controller (main/query-support) #270

Closed

9 tasks

Copilot AI and others added 2 commits April 3, 2026 08:22

fix: restore interrupt status in InterruptException catch blocks

f14aa34

Agent-Logs-Url: https://github.com/elasticsoftwarefoundation/akces-framework/sessions/6348079c-1a83-4024-90f0-bef5784778b6 Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com>

Copilot AI changed the title ~~[WIP] Implement partition and controller for Reflector feature~~ feat(query-support): Reflector Phase 3 — ReflectorPartition, ReflectorPartitionCommandBus, AkcesReflectorController Apr 3, 2026

Copilot finished work on behalf of jwijgerd April 3, 2026 08:27

Copilot AI requested a review from jwijgerd April 3, 2026 08:27

jwijgerd marked this pull request as ready for review April 3, 2026 08:54

Copilot AI review requested due to automatic review settings April 3, 2026 08:54

Copilot started reviewing on behalf of jwijgerd April 3, 2026 08:55 View session

Copilot AI reviewed Apr 3, 2026

View reviewed changes

jwijgerd approved these changes Apr 3, 2026

View reviewed changes

jwijgerd merged commit fd69e4c into main Apr 3, 2026
11 checks passed

jwijgerd deleted the copilot/reflector-feature-phase-3-partition-controller branch April 3, 2026 09:36

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat(query-support): Reflector Phase 3 — ReflectorPartition, ReflectorPartitionCommandBus, AkcesReflectorController#276

feat(query-support): Reflector Phase 3 — ReflectorPartition, ReflectorPartitionCommandBus, AkcesReflectorController#276
jwijgerd merged 3 commits into
mainfrom
copilot/reflector-feature-phase-3-partition-controller

Copilot AI commented Apr 3, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Copilot AI Apr 3, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		logger.error("Error beginning transaction for event '{}' on partition {} of {}Reflector",
		domainEventRecord.name(), topicPartition, runtime.getName(), e);

-            logger.error("Error beginning transaction for event '{}' on partition {} of {}Reflector",
-                    domainEventRecord.name(), topicPartition, runtime.getName(), e);
+            consumer.seek(topicPartition, offset);
+            logger.error("Error beginning transaction for event '{}' on partition {} of {}Reflector; seeking back to offset {} for retry",
+                    domainEventRecord.name(), topicPartition, runtime.getName(), offset, e);

	return Math.abs(hashFunction.hashString(aggregateId, UTF_8).asInt()) % partitions;
	int hash = hashFunction.hashString(aggregateId, UTF_8).asInt();
	return Math.floorMod(hash, partitions);

Uh oh!

Conversation

Copilot AI commented Apr 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

New Components

Enums

ReflectorPartitionCommandBus

ReflectorPartition

AkcesReflectorController

Notable Details

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Copilot AI commented Apr 3, 2026 •

edited

Loading

`ReflectorPartitionCommandBus`

`ReflectorPartition`

`AkcesReflectorController`