Add a distributed query bus for non-axonserver modes (issue #613)
Hi all,
Past weeks have been a bit hectic, being on vacation, came back with covid, and needing to prepare for a talk. So sorry for the late response.
Thanks a lot for all the suggestions. More are always welcome, some of them were already on our own list, it’s nice to see them validated. I’m a bit scared to make promises, but it’s almost certain we move to Java 17, and might use some a the enhancement to Java to improve the API.
There are also some nice non breaking ideas in here we might pick up later (or even earlier if time allows). As it’s open source please let us know if you would like to work on some of the things suggested. Also feel free to open a new topic if you would like to discuss some ideas in more detail.
Communicating across multiple contexts in Axon is currently unintuitive. I have struggled to find anything in the reference guide and the offered APIs are not very explicit. Since communication between multiple bounded contexts is a central concept in DDD I feel like it should be supported better.
I created two threads (Thread 1, Thread 2) on this issue and the answers helped me solve my problem. I added some thoughts on how I would imagine the API. Maybe it has some place in Axon 5. Anyways, I hope I can spark a conversation in your team on this topic.
PS: Any best practices regarding communication between contexts in Axon are more than welcome. I am still experimenting with this and I am looking for design guidelines.
1 Like
We are already supporting Springboot, but there might be things we can improve?
Yes, we plan to move to Springboot 3.
1 Like
I have three suggestions:
In order to make use of ACID transations and avoid dual writes it might be beneficial to support something like the Outbox pattern.
I already started some discussions some years ago:
- The outbox pattern - #10 by Oliver_Libutzki
- Consistency gap in conjunction with state-stored aggregates · Issue #1103 · AxonFramework/AxonFramework · GitHub
Somehow related, but with another focus:
We would like to have a native option to decide if a certain event should be exposed to the outside world or should only be visible and usable within a certain module / bounded context.
@Simon_Zambrovski, @Steven_van_Beelen and I already discussed this topic in the Contributions section. Although there has not been any action in the last couple of months the topic is still relevant for us. (For those who are allowed to see the thread: https://discuss.axoniq.io/t/local-event-store-support/2980)
From a conceptual point of view the append-only Event Store is a great idea. Unfortunately in our projects we experienced that this leads to problems for us. Sometimes it’s not feasable to extend the upcaster chain in order to evolve our events. In the context of GDPR we sometimes wish there was a solution to anonymize data in old events.
I know that “changing” the event store and therefore the history comes with its own flaws, but I just would like to share that having an append-only event store sometimes leads to cumbersome solutions and rewriting history in certain cases is much more pragmatic.
That being said we highly appreciate your work and there are a lot of ongoing improvements in the 4.x stream.
Especially the “Dead letter queue” is something that I would have requested for 5.x if it would not be planned for 4.x.
4 Likes
I’ve run into an issue today where an enahncement to the injection of resources into event handlers would be helpful. It would be really helpful to have the ability to inject attributes / fields from an event message into the event handler and have this taken into account when choosing which event handler to invoke. Something like the following:-
class MyEvent {
String attribute1;
String attribute2;
}
class AggregateClass {
...
@EventSourcingHandler(payloadType = MyEvent.class)
void on(@EventAttribute(required = false) String attribute1) {
...
}
@EventSourcingHandler(payloadType = MyEvent.class)
void on(@EventAttribute(required = true) String attribute2) {
...
}
}
In this instance I’ve made up an @EventAttribute annotation but it could be anything that makes the most sense, possibly @PayloadAttribute or @MessageAttribute
2 Likes
In a similar vein to the previous suggestion, another nice feature would be the ability to inject resources into aggregates at the time of creation, instead of having to pass resources to @ComandHandler and @EventSourcingHandler methods. This would work the same as it does for sagas, e.g.
@Component
class MyResource {
...
}
@Aggregate
class AggregateClass {
...
private transient MyResource myResource;
@Autowired
void setMyResource(MyResource myResource) {
this.myResource = myResource;
}
...
}
In this way, if the resource is used in multiple command / event handlers, it only needs to be injected once as opposed to on every method that requires it.
2 Likes
Hello @andye, I actually really like the suggestion. I do want to note that this is currently already possible! You can write this yourself and register the ParameterResolverFactory.
If we define the following annotation:
@Documented
@Target({ElementType.PARAMETER})
@Retention(RetentionPolicy.RUNTIME)
public @interface PayloadAttribute {
String path();
boolean required() default false;
}
We can then create the following factory, assuming you use Jackson:
public class PayloadAttributeParameterResolverFactory implements ParameterResolverFactory {
@Override
public ParameterResolver createInstance(final Executable executable, final Parameter[] parameters, final int i) {
if (AnnotationUtils.isAnnotationPresent(parameters[i], PayloadAttribute.class)) {
final Map<String, Object> propertyMap = AnnotationUtils.findAnnotationAttributes(parameters[i], PayloadAttribute.class).get();
return new PayloadAttributeParameterResolver((String) propertyMap.get("path"), (Boolean) propertyMap.get("required"));
}
return null;
}
private static class PayloadAttributeParameterResolver implements ParameterResolver<Object> {
private final ObjectMapper objectMapper = new ObjectMapper().findAndRegisterModules();
private final String path;
private final boolean required;
PayloadAttributeParameterResolver(String path, boolean required) {
this.path = path;
this.required = required;
}
@Override
public Object resolveParameterValue(final Message<?> message) {
JsonNode currentNode = objectMapper.valueToTree(message.getPayload());
for (String part : path.split("\\.")) {
if (!currentNode.isObject() || !currentNode.has(part) || currentNode.isNull()) {
currentNode = null;
break;
}
currentNode = currentNode.get(part);
}
if (currentNode == null || currentNode.isNull()) {
return null;
}
if (currentNode.isTextual()) {
return currentNode.textValue();
}
if (currentNode.isDouble()) {
return currentNode.doubleValue();
}
if (currentNode.isLong()) {
return currentNode.longValue();
}
if(currentNode.isInt()) {
return currentNode.intValue();
}
return null;
}
@Override
public boolean matches(final Message<?> message) {
if (!required) {
return true;
}
return resolveParameterValue(message) != null;
}
}
}
This will provide the functionality to use multiple methods for the same event based on payload, for example:
@Aggregate
class Account {
private Account() {
}
@CommandHandler
Account(RegisterAccountCommand command, @PayloadAttribute(path = "password", required = true) String password) {
apply(new AccountRegisteredEvent(command.getAccountId(), command.getUserName(), command.getPassword()));
}
@CommandHandler
Account(RegisterAccountCommand command) {
apply(new AccountRegisteredEvent(command.getAccountId(), command.getUserName(), "nopassword"));
}
}
This sample uses the annotation to deviate between a command with and without a password, but you can use it any way you like!
1 Like
With Java 19 Virtual Threads (JEP 425) are added as an experimental feature. As Axon Framewors deals a lot with threads, especially the Streaming Event Processors, it might be interesting to support Virtual Threads.
I am aware of the PooledSteamingEventProcessor which uses ThreadPools to reduce the number of concurrent threads. With Virtual Threads using a thread pool doesn’t make much sense as those threads are lightweight as they are not bound to an OS thread. Not sure, but maybe the good old TrackingEventProcessor could get some traction again.
I don’t not which JDK baseline you are targeting for with Axon Framework 5, but I guess it’s likely that it won’t be Java >= 19. As a consequence you can’t use the Virtual Thread API, but maybe you can provide a way / an API to configure it, so Java 19 applications can benefit from virtuals threads.
1 Like
Virtual threads are interesting indeed. It might be a small extension or something I think. Not sure that’s even needed since they should be rather compatible with normal Threads. But they are cheaper, so it might make since to prefer TEP over PSEP.
Hello @Oliver_Libutzki,
After some experimentation, I found that we are already compatible with Virtual Threads. The TEP uses a ThreadFactory, which you can configure, like so:
config.registerTrackingEventProcessorConfiguration(configuration ->
TrackingEventProcessorConfiguration.forParallelProcessing(4)
.andInitialSegmentsCount(4)
.andThreadFactory(s -> Thread
.ofVirtual()
.allowSetThreadLocals(true)
.inheritInheritableThreadLocals(false)
.name(s + "-", 0).factory()
)
);
Please let me know if you will use it and how it turns out! Locally it seems to work pretty well.
Kind regards,
Mitchell Herrijgers
1 Like
Based on my limited reading of virtual threads I foresee the following challenges in making use of them to their fullest with Axon:
- Usage of
synchronizedwithin Axon and Spring-
From this thread it is clear that it is technically possible to remove this barrier but not a priority for JDK developers at the moment:
It requires rewriting monitors in Java, and moving any part of the runtime to Java is complicated (e.g. there are interesting interactions with JVMTI, that normally assumes that all Java code is user code). Also, it isn’t a very high priority, as library developers seem happy to remove pinning
-
So the solution is to either change from using
synchronizedto using locks in every place in Axon and Spring or just wait until they implement the feature - otherwise the virtual thread will just pin anyway when it comes acrosssynchronized.
-
- Usage of thread locals
- The way they seem to be going is to use “extent-local” variables instead.
- See JEP 429 Extent-local variables
- Essentially the data is immutable so easily shareable.
- Extremely large stacks may cause problems
- The trick here would be to spawn the virtual thread near the top of the stack for a logical task rather than having it create a huge call stack. Spring probably has some cleaning up to do here.
Those are my observations as a casual user. You could probably get further than the current state by using the config given by @Morlack 
3 Likes
Removal of Xstream dependency from axon framework as it has too many vulnerabilities.
Use of google protocol buffers for serialization
@Ashwini_Kumar We actually thinking of leaving XStream as the default for serialisation. Not sure what would be a good other default, or maybe it would be good not to have a default. I do like Protobuf, but it has some downsides too. Please note that the Framework is open enough to use Protobuf currently.
this works fine for events, but query handlers are still run on platform threads, is there a similar way to set thread factory for query processing?
I would like to see Avro as serialization format for messages. Especially, the serialization of the metadata and message body should be not two unrelated invocations, but I want to be able to use metadata during deserialization of the message payload.
2 Likes
Can we get a Saga serialization store based on Axon Server?
Having support for snapshots by just serializing the aggregate I see no qualitative difference to storage of Sagas by serializing them too.
This would make it easier to build applications which already have Axon Server, since no additional database is required, just for storage of serialization of Sagas. Sometimes the DB is already at place, but if not (especially for the command side) it seems silly to maintain a DB (and backup and management, and ops) of a DB just to store saga state.
1 Like