From 70368005c5d97ae93e378254387aee383bfcd937 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 4 Jan 2026 11:23:05 +0000 Subject: [PATCH 1/3] Initial plan From c1ae60930adbd2f2c4a6e786617d64c1530c2ab8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 4 Jan 2026 11:30:13 +0000 Subject: [PATCH 2/3] Add AggregateNotFoundError support to framework Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../AggregateBeanFactoryPostProcessor.java | 7 +++- .../akces/kafka/KafkaAggregateRuntime.java | 35 +++++++++++++++++++ .../errors/AggregateNotFoundErrorEvent.java | 33 +++++++++++++++++ 3 files changed, 74 insertions(+), 1 deletion(-) create mode 100644 main/shared/src/main/java/org/elasticsoftware/akces/errors/AggregateNotFoundErrorEvent.java diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateBeanFactoryPostProcessor.java b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateBeanFactoryPostProcessor.java index ba940b2c..964ff37c 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateBeanFactoryPostProcessor.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateBeanFactoryPostProcessor.java @@ -24,6 +24,7 @@ import org.elasticsoftware.akces.commands.Command; import org.elasticsoftware.akces.commands.CommandBus; import org.elasticsoftware.akces.errors.AggregateAlreadyExistsErrorEvent; +import org.elasticsoftware.akces.errors.AggregateNotFoundErrorEvent; import org.elasticsoftware.akces.errors.CommandExecutionErrorEvent; import org.elasticsoftware.akces.events.DomainEvent; import org.elasticsoftware.akces.events.ErrorEvent; @@ -57,11 +58,15 @@ public class AggregateBeanFactoryPostProcessor implements BeanFactoryPostProcess new DomainEventType<>("CommandExecutionError", 1, CommandExecutionErrorEvent.class, false, false, true, false) ); public static final List> COMMAND_HANDLER_SYSTEM_ERRORS = List.of( + new DomainEventType<>("AggregateNotFoundError", 1, AggregateNotFoundErrorEvent.class, false, false, true, false), new DomainEventType<>("CommandExecutionError", 1, CommandExecutionErrorEvent.class, false, false, true, false) ); public static final List> EVENT_HANDLER_CREATE_SYSTEM_ERRORS = List.of( new DomainEventType<>("AggregateAlreadyExistsError", 1, AggregateAlreadyExistsErrorEvent.class, false, false, true, false) ); + public static final List> EVENT_HANDLER_SYSTEM_ERRORS = List.of( + new DomainEventType<>("AggregateNotFoundError", 1, AggregateNotFoundErrorEvent.class, false, false, true, false) + ); @Override @SuppressWarnings("unchecked") @@ -469,7 +474,7 @@ private List> generateDomainEventTypes(Class> generateEventHandlerErrorEventTypes(Class[] domainEventClasses, boolean isCreate) { - Stream> systemErrorEvents = (isCreate) ? EVENT_HANDLER_CREATE_SYSTEM_ERRORS.stream() : Stream.empty(); + Stream> systemErrorEvents = (isCreate) ? EVENT_HANDLER_CREATE_SYSTEM_ERRORS.stream() : EVENT_HANDLER_SYSTEM_ERRORS.stream(); return Stream.concat(Arrays.stream(domainEventClasses).map(eventClass -> { DomainEventInfo eventInfo = eventClass.getAnnotation(DomainEventInfo.class); return new DomainEventType<>(eventInfo.type(), eventInfo.version(), eventClass, false, false, true, hasPIIDataAnnotation(eventClass)); diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/KafkaAggregateRuntime.java b/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/KafkaAggregateRuntime.java index 9e6b7461..691ed859 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/KafkaAggregateRuntime.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/KafkaAggregateRuntime.java @@ -27,6 +27,7 @@ import org.elasticsoftware.akces.commands.Command; import org.elasticsoftware.akces.commands.CommandBus; import org.elasticsoftware.akces.errors.AggregateAlreadyExistsErrorEvent; +import org.elasticsoftware.akces.errors.AggregateNotFoundErrorEvent; import org.elasticsoftware.akces.errors.CommandExecutionErrorEvent; import org.elasticsoftware.akces.events.DomainEvent; import org.elasticsoftware.akces.events.ErrorEvent; @@ -161,6 +162,22 @@ private void aggregateAlreadyExists(ProtocolRecord commandOrDomainEventRecord, protocolRecordConsumer.accept(eventRecord); } + private void aggregateNotFound(ProtocolRecord commandOrDomainEventRecord, + Consumer protocolRecordConsumer) { + AggregateNotFoundErrorEvent errorEvent = new AggregateNotFoundErrorEvent(commandOrDomainEventRecord.aggregateId(), this.getName()); + DomainEventType type = getDomainEventType(errorEvent.getClass()); + DomainEventRecord eventRecord = new DomainEventRecord( + commandOrDomainEventRecord.tenantId(), + type.typeName(), + type.version(), + serialize(errorEvent), + getEncoding(type), + errorEvent.getAggregateId(), + commandOrDomainEventRecord.correlationId(), + -1L); // ErrorEvents have no generation number because they don't alter the state + protocolRecordConsumer.accept(eventRecord); + } + private void commandExecutionError(CommandRecord commandRecord, Consumer protocolRecordConsumer, Throwable exception) { @@ -265,6 +282,15 @@ private void handleCommand(CommandType commandType, // does have PIIData. in that case we need to load the proper GDPRContext Command command = materialize(commandType, commandRecord); AggregateStateRecord currentStateRecord = stateRecordSupplier.get(); + // if we don't have state, this is an error and the aggregate does not exist + if (currentStateRecord == null) { + log.warn("Command {} wants to update a {} Aggregate with id {}, but the state does not exist. Generating a AggregateNotFoundError", + commandRecord.name(), + getName(), + commandRecord.aggregateId()); + aggregateNotFound(commandRecord, protocolRecordConsumer); + return; + } AggregateState currentState = materialize(currentStateRecord); Stream domainEvents = commandHandlers.get(commandType).apply(command, currentState); for (DomainEvent domainEvent : domainEvents.toList()) @@ -333,6 +359,15 @@ private void handleEvent(DomainEventType eventType, // materialize the event DomainEvent externalEvent = materialize(eventType, domainEventRecord); AggregateStateRecord currentStateRecord = stateRecordSupplier.get(); + // if we don't have state, this is an error and the aggregate does not exist + if (currentStateRecord == null) { + log.warn("External DomainEvent {} wants to update a {} Aggregate with id {}, but the state does not exist. Generating a AggregateNotFoundError", + domainEventRecord.name(), + getName(), + domainEventRecord.aggregateId()); + aggregateNotFound(domainEventRecord, protocolRecordConsumer); + return; + } AggregateState currentState = materialize(currentStateRecord); Stream domainEvents = eventHandlers.get(eventType).apply(externalEvent, currentState); for (DomainEvent domainEvent : domainEvents.toList()) diff --git a/main/shared/src/main/java/org/elasticsoftware/akces/errors/AggregateNotFoundErrorEvent.java b/main/shared/src/main/java/org/elasticsoftware/akces/errors/AggregateNotFoundErrorEvent.java new file mode 100644 index 00000000..16de9625 --- /dev/null +++ b/main/shared/src/main/java/org/elasticsoftware/akces/errors/AggregateNotFoundErrorEvent.java @@ -0,0 +1,33 @@ +/* + * Copyright 2022 - 2026 The Original Authors + * + * Licensed 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.elasticsoftware.akces.errors; + +import jakarta.validation.constraints.NotNull; +import org.elasticsoftware.akces.annotations.DomainEventInfo; +import org.elasticsoftware.akces.events.ErrorEvent; + +@DomainEventInfo(type = "AggregateNotFoundError") +public record AggregateNotFoundErrorEvent( + @NotNull String aggregateId, + @NotNull String aggregateName +) implements ErrorEvent { + @Override + public String getAggregateId() { + return aggregateId(); + } +} From a7c338bf1ad6dddee704c00d1a60a7567b469420 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 4 Jan 2026 11:35:43 +0000 Subject: [PATCH 3/3] Add test for AggregateNotFoundError functionality Co-authored-by: jwijgerd <914840+jwijgerd@users.noreply.github.com> --- .../akcestest/RuntimeTests.java | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/RuntimeTests.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/RuntimeTests.java index b257ea7c..ec55a12c 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/RuntimeTests.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/RuntimeTests.java @@ -48,6 +48,7 @@ import org.elasticsoftware.akces.control.AggregateServiceRecord; import org.elasticsoftware.akces.control.AkcesControlRecord; import org.elasticsoftware.akces.errors.AggregateAlreadyExistsErrorEvent; +import org.elasticsoftware.akces.errors.AggregateNotFoundErrorEvent; import org.elasticsoftware.akces.events.DomainEvent; import org.elasticsoftware.akces.gdpr.jackson.AkcesGDPRModule; import org.elasticsoftware.akces.protocol.*; @@ -881,6 +882,29 @@ public void testAggregateAlreadyExistsErrorWithAkcesClient() throws ExecutionExc } + @Test + @Order(12) + public void testAggregateNotFoundErrorWithAkcesClient() throws ExecutionException, InterruptedException, TimeoutException { + // wait until the akces controller is running + while (!walletAggregateController.isRunning() || + !accountAggregateController.isRunning() || + !orderProcessManagerAggregateController.isRunning() || + !akcesClient.isRunning()) { + Thread.onSpinWait(); + } + + // Generate a random wallet ID that doesn't exist + String nonExistentWalletId = UUID.randomUUID().toString(); + + // Try to credit a wallet that doesn't exist + CreditWalletCommand command = new CreditWalletCommand(nonExistentWalletId, "USD", new BigDecimal("100.00")); + List result = akcesClient.send("TEST_TENANT", command).toCompletableFuture().get(10, TimeUnit.SECONDS); + + Assertions.assertNotNull(result); + Assertions.assertEquals(1, result.size()); + assertInstanceOf(AggregateNotFoundErrorEvent.class, result.getFirst()); + } + public TopicDescription getTopicDescription(String topic) { try { return adminClient.describeTopics(topic).get(topic);