Support customize the KafkaMessageSender used by DelegatingEventExternalizer

[CrisLi]

In our project, we need to do some decoration and a specific message serializer to send out the Kafka message. Our current code is we use a delegated KafkaTemplate.

We want to involve the Spring Modulith Events into our project to implement an outbox pattern. But we don't want to break any of the existing Kafka message features. So is there a way we can customize the DelegatingEventExternalizer? Currently, there is no conditional annotation on the DelegatingEventExternalizer in the KafkaEventExternalizerConfiguration

	@Bean
	DelegatingEventExternalizer kafkaEventExternalizer(EventExternalizationConfiguration configuration,
			KafkaOperations<Object, Object> operations, BeanFactory factory) {

		logger.debug("Registering domain event externalization to Kafka…");

		var context = new StandardEvaluationContext();
		context.setBeanResolver(new BeanFactoryResolver(factory));

		return new DelegatingEventExternalizer(configuration, (target, payload) -> {

			var routing = BrokerRouting.of(target, context);
			return operations.send(routing.getTarget(), routing.getKey(payload), payload);
		});
	}

[odrotbohm]

Can you elaborate on what kind of customizations you envision? There's currently no opt-out because all the spring-modulith-events-kafka ships is the declaration. We just recently added support to add headers to the message sent out, but any kind of further customization brings you closer to simply depending on spring-modulith-events-core and declaring a custom bean.

[CrisLi]

My case is that we need to sign the external event playload (Kafka message value) with a private key, generate a signature and put the signature into the Kakfa message header.

In the consumer side, our customized Kakfa consumers would read the signature from the header, verify the message playload with a public key.

Since we have so much legacy systems which already be implemented as this model. We have to make sure the new project systems send the Kafka message in the same way.

Now, I found a tradeoff solution, we define a @primary KafkaOperation bean and set a KafkaProducerIntercepter to it. We do our logic in the intercepter.

The disadvantage of this solution is we have to serialize the message value to json twice. One is in the intercepter and one in the final serializer. Because we need to sign the message value before send it.

If we can add a @ConditionalOnMissingBean to DelegatingEventExternalizer bean factory method. Then we would have chance to do some things before sending the "object" to Kafka.

[odrotbohm]

Would you mind giving the new header augmentation feature of the upcoming 1.3 release a try? EventExternalizationConfiguration exposes API to produce a header map per event object to be included with the message sent out to the broker.

Although we could register DelegatingEventExternalizer more defensively, I'm not quite ready to do that. As of now, it has not really been designed to be considered API and registering with a @ConditionalOnMissingBean would kind of do that. If in doubt, I still think you're better off registering a custom @ApplicationModuleListener (make sure to disable transactions for that), get EventExternalizationConfiguration and KafkaTemplate injected and take it from there.