Jackson fails to serialize some events for Kafka

n3ziniuka5 · October 17, 2023, 9:49am

Hello,

I wanted to add Kafka extension to an existing Axon project. Here’s my configuration:

axon:
  serializer:
    general: jackson
  axonserver:
    servers: ${AXONSERVER_SERVERS:localhost}
    token: ${AXON_APPLICATION_TOKEN:''}
  kafka:
    bootstrap-servers: ${KAFKA_URL:localhost:9092}
    client-id: axon-kafka-test
    default-topic: ${KAFKA_TOPIC:test}
    properties:
      security.protocol: PLAINTEXT

    publisher:
      confirmation-mode: transactional

    producer:
      transaction-id-prefix: axon-kafka-test
      event-processor-mode: tracking

Everything worked fine locally. However, once I deployed it threw this exception:

org.axonframework.serialization.SerializationException: Unable to serialize object
	at org.axonframework.serialization.json.JacksonSerializer.serialize(JacksonSerializer.java:149) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.serialization.SerializedMessage.serializePayload(SerializedMessage.java:123) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.messaging.MessageDecorator.serializePayload(MessageDecorator.java:66) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.extensions.kafka.eventhandling.DefaultKafkaMessageConverter.createKafkaMessage(DefaultKafkaMessageConverter.java:126) ~[axon-kafka-4.8.0.jar:4.8.0]
	at org.axonframework.extensions.kafka.eventhandling.producer.KafkaPublisher.send(KafkaPublisher.java:151) ~[axon-kafka-4.8.0.jar:4.8.0]
	at org.axonframework.extensions.kafka.eventhandling.producer.KafkaEventPublisher.handle(KafkaEventPublisher.java:77) ~[axon-kafka-4.8.0.jar:4.8.0]
	at org.axonframework.eventhandling.SimpleEventHandlerInvoker.invokeHandlers(SimpleEventHandlerInvoker.java:128) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.SimpleEventHandlerInvoker.handle(SimpleEventHandlerInvoker.java:114) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.MultiEventHandlerInvoker.handle(MultiEventHandlerInvoker.java:91) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.AbstractEventProcessor.lambda$null$1(AbstractEventProcessor.java:174) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.tracing.Span.runCallable(Span.java:132) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.AbstractEventProcessor.lambda$null$2(AbstractEventProcessor.java:171) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.messaging.DefaultInterceptorChain.proceed(DefaultInterceptorChain.java:57) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.messaging.interceptors.CorrelationDataInterceptor.handle(CorrelationDataInterceptor.java:67) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.messaging.DefaultInterceptorChain.proceed(DefaultInterceptorChain.java:55) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.tracing.Span.runCallable(Span.java:132) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.TrackingEventProcessor.lambda$new$3(TrackingEventProcessor.java:185) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.messaging.DefaultInterceptorChain.proceed(DefaultInterceptorChain.java:55) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.TrackingEventProcessor.lambda$new$1(TrackingEventProcessor.java:179) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.messaging.DefaultInterceptorChain.proceed(DefaultInterceptorChain.java:55) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.AbstractEventProcessor.lambda$processInUnitOfWork$3(AbstractEventProcessor.java:182) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.messaging.unitofwork.BatchingUnitOfWork.executeWithResult(BatchingUnitOfWork.java:92) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.AbstractEventProcessor.processInUnitOfWork(AbstractEventProcessor.java:165) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.TrackingEventProcessor.processBatch(TrackingEventProcessor.java:490) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.TrackingEventProcessor.processingLoop(TrackingEventProcessor.java:318) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.TrackingEventProcessor$TrackingSegmentWorker.run(TrackingEventProcessor.java:1145) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.TrackingEventProcessor$WorkerLauncher.cleanUp(TrackingEventProcessor.java:1340) ~[axon-messaging-4.8.1.jar:4.8.1]
	at org.axonframework.eventhandling.TrackingEventProcessor$WorkerLauncher.run(TrackingEventProcessor.java:1317) ~[axon-messaging-4.8.1.jar:4.8.1]
	at java.base/java.lang.Thread.run(Unknown Source) ~[na:na]
Caused by: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: No serializer found for class org.axonframework.serialization.UnknownSerializedType and no properties discovered to create BeanSerializer (to avoid exception, disable SerializationFeature.FAIL_ON_EMPTY_BEANS)
	at com.fasterxml.jackson.databind.exc.InvalidDefinitionException.from(InvalidDefinitionException.java:77) ~[jackson-databind-2.14.2.jar:2.14.2]
	at com.fasterxml.jackson.databind.SerializerProvider.reportBadDefinition(SerializerProvider.java:1306) ~[jackson-databind-2.14.2.jar:2.14.2]
	at com.fasterxml.jackson.databind.DatabindContext.reportBadDefinition(DatabindContext.java:408) ~[jackson-databind-2.14.2.jar:2.14.2]
	at com.fasterxml.jackson.databind.ser.impl.UnknownSerializer.failForEmpty(UnknownSerializer.java:53) ~[jackson-databind-2.14.2.jar:2.14.2]
	at com.fasterxml.jackson.databind.ser.impl.UnknownSerializer.serialize(UnknownSerializer.java:30) ~[jackson-databind-2.14.2.jar:2.14.2]
	at com.fasterxml.jackson.databind.ser.DefaultSerializerProvider._serialize(DefaultSerializerProvider.java:480) ~[jackson-databind-2.14.2.jar:2.14.2]
	at com.fasterxml.jackson.databind.ser.DefaultSerializerProvider.serializeValue(DefaultSerializerProvider.java:319) ~[jackson-databind-2.14.2.jar:2.14.2]
	at com.fasterxml.jackson.databind.ObjectWriter$Prefetch.serialize(ObjectWriter.java:1572) ~[jackson-databind-2.14.2.jar:2.14.2]
	at com.fasterxml.jackson.databind.ObjectWriter._writeValueAndClose(ObjectWriter.java:1273) ~[jackson-databind-2.14.2.jar:2.14.2]
	at com.fasterxml.jackson.databind.ObjectWriter.writeValueAsBytes(ObjectWriter.java:1163) ~[jackson-databind-2.14.2.jar:2.14.2]
	at org.axonframework.serialization.json.JacksonSerializer.serialize(JacksonSerializer.java:144) ~[axon-messaging-4.8.1.jar:4.8.1]
	... 28 common frames omitted

What does this error mean exactly and how should I handle this situation? As I understand it, it may be that we’ve deleted the data class of some very old event that now can’t be serialized. It would be helpful if the exception contained the actual class it was looking for. So should I implement a serializer for org.axonframework.serialization.UnknownSerializedType or disable SerializationFeature.FAIL_ON_EMPTY_BEANS (and how would I disable it)?

n3ziniuka5 · October 17, 2023, 10:43am

We’ve replayed some older projections and they processed fine, so it can’t be that we deleted some classes for older events.

Gerard · October 17, 2023, 6:41pm

An event would unlikely be empty. It’s likely easier to find the event causing the problem in Axon Server. Events are only serialized if needed, so it still could be some class not part of the projections maybe?

n3ziniuka5 · October 17, 2023, 7:04pm

We have multiple applications connected to Axon server. I was under the impression that I needed to add a Kafka producer to each of them. However, it turned out that with tracking processor mode it processes every event that entered the Axon server, even if it wasn’t produced by that application.

So the error makes sense, it tried processing an event that didn’t have a corresponding class on the classpath, and the serializer failed.

I am evaluating what to do to make the event processor only process events created by that same application. I’d prefer not to use subscribing event processor mode for added reliability in case Kafka is down. Also, the Kafka extension doesn’t handle transaction timeouts correctly (if a transaction commit times out, you cannot rollback because the transactions may have already been committed. You must keep retrying until you get confirmation from the broker.). Other confirmation modes make it possible for an event to be present in Kafka, but not in Axon so they are also not an option.

Do you have any suggestions? I can explore the events on the Axon server UI, so they must be persisted as Json, is it possible to have Kafka extension simply publish that raw json instead of trying to use a serializer with event upcasting chain?

Gerard · October 17, 2023, 7:30pm

Sounds like it might be better/easier to not use the Kafka extension, and have a regular event handler which sends some events to Kafka.

If you just want to filter certain events, you can configure a topic resolver. To change the serialization when publishing to Kafka, you can set a specific serializer.

Personally I don’t really see the value of Kafka transactions for this use case. In my view it’s better to expect duplications on the Kafka side, and make sure only to update the token store when you are sure it’s properly received. But not to strive for exactly once, since that’s impossible anyway.

n3ziniuka5 · October 31, 2023, 9:39am

With the release of 4.9.0 it’s now very easy to configure a custom topic resolver, so that solves my issue.