feature: add upcasting functionality for DomainEvent and AggregateState classes

main/api/src/main/java/org/elasticsoftware/akces/aggregate/AggregateStateType.java

@@ -25,5 +25,5 @@ public record AggregateStateType<C extends AggregateState>(
         boolean indexed,
         String indexName,
         boolean piiData
-) {
+) implements ProtocolRecordType<C> {
 }

main/api/src/main/java/org/elasticsoftware/akces/aggregate/CommandType.java

@@ -27,7 +27,7 @@ public record CommandType<C extends Command>(
         boolean create,
         boolean external,
         boolean piiData
-) implements SchemaType {
+) implements SchemaType<C> {
 
     @Override
     public String getSchemaPrefix() {

main/api/src/main/java/org/elasticsoftware/akces/aggregate/DomainEventType.java

@@ -28,7 +28,7 @@ public record DomainEventType<T extends DomainEvent>(
         boolean external,
         boolean error,
         boolean piiData
-) implements SchemaType {
+) implements SchemaType<T> {
 
     @Override
     public String getSchemaPrefix() {

main/api/src/main/java/org/elasticsoftware/akces/aggregate/ProtocolRecordType.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2022 - 2025 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.aggregate;
+
+public interface ProtocolRecordType<T> {
+    String typeName();
+
+    int version();
+
+    Class<T> typeClass();
+}

main/api/src/main/java/org/elasticsoftware/akces/aggregate/SchemaType.java

@@ -17,7 +17,7 @@
 
 package org.elasticsoftware.akces.aggregate;
 
-public interface SchemaType {
+public interface SchemaType<T> extends ProtocolRecordType<T> {
     String getSchemaPrefix();
 
     default String getSchemaName() {
@@ -26,11 +26,5 @@ default String getSchemaName() {
 
     boolean relaxExternalValidation();
 
-    String typeName();
-
-    int version();
-
-    Class<?> typeClass();
-
     boolean external();
 }

main/api/src/main/java/org/elasticsoftware/akces/aggregate/UpcastingHandlerFunction.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2022 - 2025 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.aggregate;
+
+@FunctionalInterface
+public interface UpcastingHandlerFunction<T , R , TR extends ProtocolRecordType<T>, RR extends ProtocolRecordType<R>> {
+    R apply(T t);
+
+    default TR getInputType() {
+        throw new UnsupportedOperationException("When implementing UpcastingHandlerFunction directly, you must override getInputType()");
+    }
+
+    default RR getOutputType() {
+        throw new UnsupportedOperationException("When implementing UpcastingHandlerFunction directly, you must override getOutputType()");
+    }
+
+    default Aggregate<? extends AggregateState> getAggregate() {
+        throw new UnsupportedOperationException("When implementing UpcastingHandlerFunction directly, you must override getAggregate()");
+    }
+}

main/api/src/main/java/org/elasticsoftware/akces/annotations/AggregateInfo.java

@@ -32,7 +32,7 @@
     @AliasFor(annotation = Component.class)
     String value();
 
-    int version() default 1;
+    int stateVersion() default 1;
 
     boolean generateGDPRKeyOnCreate() default false;
 

main/api/src/main/java/org/elasticsoftware/akces/annotations/AggregateStateInfo.java

@@ -17,6 +17,13 @@
 
 package org.elasticsoftware.akces.annotations;
 
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.TYPE})
 public @interface AggregateStateInfo {
     String type();
 

main/api/src/main/java/org/elasticsoftware/akces/annotations/CommandInfo.java

@@ -17,10 +17,13 @@
 
 package org.elasticsoftware.akces.annotations;
 
+import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
 
 @Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.TYPE})
 public @interface CommandInfo {
     String type();
 

main/api/src/main/java/org/elasticsoftware/akces/annotations/DomainEventInfo.java

@@ -17,10 +17,13 @@
 
 package org.elasticsoftware.akces.annotations;
 
+import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
 
 @Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.TYPE})
 public @interface DomainEventInfo {
     String type();
 

main/api/src/main/java/org/elasticsoftware/akces/annotations/UpcastingHandler.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2022 - 2025 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.annotations;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target({ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface UpcastingHandler {
+}

main/api/src/main/java/org/elasticsoftware/akces/commands/CommandBusHolder.java

@@ -25,6 +25,7 @@ public class CommandBusHolder {
     protected CommandBusHolder() {
     }
 
+    @SuppressWarnings("rawtypes")
     public static CommandBus getCommandBus(Class<? extends Aggregate> aggregateClass) {
         return commandBusThreadLocal.get();
     }

main/client/src/test/java/org/elasticsoftware/akces/client/AkcesClientTests.java

@@ -120,6 +120,7 @@ public class AkcesClientTests {
                     .withNetwork(network)
                     .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "kafka:9092")
                     .withEnv("SCHEMA_REGISTRY_HOST_NAME", "localhost")
+                    .withEnv("SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL","none")
                     .withExposedPorts(8081)
                     .withNetworkAliases("schema-registry")
                     .dependsOn(kafka);

main/query-support/src/main/java/org/elasticsoftware/akces/query/database/beans/DatabaseModelEventHandlerFunctionAdapter.java

@@ -1,21 +1,21 @@
 /*
- * Copyright 2022 - 2025 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.
- *
- */
+  * Copyright 2022 - 2025 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.query.database.beans;
+ package org.elasticsoftware.akces.query.database.beans;
 
 import jakarta.validation.constraints.NotNull;
 import org.elasticsoftware.akces.aggregate.DomainEventType;
@@ -24,70 +24,68 @@
 import org.elasticsoftware.akces.query.DatabaseModel;
 import org.elasticsoftware.akces.query.DatabaseModelEventHandlerFunction;
 
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
 
-public class DatabaseModelEventHandlerFunctionAdapter<E extends DomainEvent>
-        implements DatabaseModelEventHandlerFunction<E> {
-    private final DatabaseModel databaseModel;
-    private final String adapterMethodName;
-    private final Class<E> domainEventClass;
-    private final DomainEventType<E> domainEventType;
-    private Method adapterMethod;
+ public class DatabaseModelEventHandlerFunctionAdapter<E extends DomainEvent>
+         implements DatabaseModelEventHandlerFunction<E> {
+     private final DatabaseModel databaseModel;
+     private final String adapterMethodName;
+     private final Class<E> domainEventClass;
+     private final DomainEventType<E> domainEventType;
+     private MethodHandle adapterMethodHandle;
 
-    public DatabaseModelEventHandlerFunctionAdapter(DatabaseModel databaseModel,
-                                                    String adapterMethodName,
-                                                    Class<E> domainEventClass,
-                                                    String typeName,
-                                                    int version) {
-        this.databaseModel = databaseModel;
-        this.adapterMethodName = adapterMethodName;
-        this.domainEventClass = domainEventClass;
-        this.domainEventType = new DomainEventType<>(
-                typeName,
-                version,
-                domainEventClass,
-                false,
-                true,
-                false,
-                GDPRAnnotationUtils.hasPIIDataAnnotation(domainEventClass));
-    }
+     public DatabaseModelEventHandlerFunctionAdapter(DatabaseModel databaseModel,
+                                                     String adapterMethodName,
+                                                     Class<E> domainEventClass,
+                                                     String typeName,
+                                                     int version) {
+         this.databaseModel = databaseModel;
+         this.adapterMethodName = adapterMethodName;
+         this.domainEventClass = domainEventClass;
+         this.domainEventType = new DomainEventType<>(
+                 typeName,
+                 version,
+                 domainEventClass,
+                 false,
+                 true,
+                 false,
+                 GDPRAnnotationUtils.hasPIIDataAnnotation(domainEventClass));
+     }
 
-    @SuppressWarnings("unused")
-    public void init() {
-        try {
-            adapterMethod = databaseModel.getClass().getMethod(adapterMethodName, domainEventClass);
-        } catch (NoSuchMethodException e) {
-            throw new RuntimeException(e);
-        }
-    }
+     @SuppressWarnings("unused")
+     public void init() {
+         try {
+             MethodHandles.Lookup lookup = MethodHandles.lookup();
+             MethodType methodType = MethodType.methodType(void.class, domainEventClass);
+             adapterMethodHandle = lookup.findVirtual(databaseModel.getClass(), adapterMethodName, methodType);
+         } catch (NoSuchMethodException | IllegalAccessException e) {
+             throw new RuntimeException("Failed to find method " + adapterMethodName + " on " +
+                     databaseModel.getClass().getName(), e);
+         }
+     }
 
-    @Override
-    public void accept(@NotNull E event) {
-        try {
-            adapterMethod.invoke(databaseModel, event);
-        } catch (IllegalAccessException e) {
-            throw new RuntimeException(e);
-        } catch (InvocationTargetException e) {
-            if (e.getCause() != null) {
-                if (e.getCause() instanceof RuntimeException) {
-                    throw (RuntimeException) e.getCause();
-                } else {
-                    throw new RuntimeException(e.getCause());
-                }
-            } else {
-                throw new RuntimeException(e);
-            }
-        }
-    }
+     @Override
+     public void accept(@NotNull E event) {
+         try {
+             adapterMethodHandle.invoke(databaseModel, event);
+         } catch (Throwable e) {
+             if (e instanceof RuntimeException) {
+                 throw (RuntimeException) e;
+             } else {
+                 throw new RuntimeException("Error invoking event handler", e);
+             }
+         }
+     }
 
-    @Override
-    public DomainEventType<E> getEventType() {
-        return domainEventType;
-    }
+     @Override
+     public DomainEventType<E> getEventType() {
+         return domainEventType;
+     }
 
-    @Override
-    public DatabaseModel getDatabaseModel() {
-        return databaseModel;
-    }
-}
+     @Override
+     public DatabaseModel getDatabaseModel() {
+         return databaseModel;
+     }
+ }

main/query-support/src/main/java/org/elasticsoftware/akces/query/models/AkcesQueryModelController.java

@@ -160,9 +160,9 @@ private void process(Consumer<String, ProtocolRecord> indexConsumer) {
                 if(!newExecutions.isEmpty()){
                     logger.info("Processing {} new HydrationExecutions", newExecutions.size());
                     // we need to get the endoffsets for all the partitions
-                    indexConsumer.endOffsets(newExecutions.keySet()).forEach((partition, endOffset) -> {
-                        hydrationExecutions.computeIfPresent(partition, (topicPartition, hydrationExecution) -> hydrationExecution.withEndOffset(endOffset));
-                    });
+                    indexConsumer.endOffsets(newExecutions.keySet()).forEach((partition, endOffset) ->
+                            hydrationExecutions.computeIfPresent(partition, (topicPartition, hydrationExecution) ->
+                                    hydrationExecution.withEndOffset(endOffset)));
                 }
                 // seek to the correct offset for the new executions
                 newExecutions.forEach((partition, execution) -> {

main/query-support/src/main/java/org/elasticsoftware/akces/query/models/QueryModelRuntimeFactory.java

@@ -65,7 +65,7 @@ private QueryModelRuntime<S> createRuntime(QueryModel<S> queryModel) {
         QueryModelInfo queryModelInfo = queryModel.getClass().getAnnotation(QueryModelInfo.class);
 
         if (queryModelInfo != null) {
-            runtimeBuilder.setStateType(new QueryModelStateType<S>(
+            runtimeBuilder.setStateType(new QueryModelStateType<>(
                     queryModelInfo.value(),
                     queryModelInfo.version(),
                     queryModel.getStateClass(),