Historical Note: This was originally written in early 2022, there may be better ways now.

This post demonstrates how to send Kafka events to dynamic topics in Red Hat Quarkus 2.

I previously used the KafkaProducer, but the metadata was ignored resulting in serialization issues.

The solution was to use a standard emitter and to attach the topic metadata directly to the outgoing message. 

@ApplicationScoped
public class RoutingReflector implements Reflector {
    private static final Logger logger = Logger.getLogger(RouteTable.class);

    @Inject
    @Channel("reflector")
    Emitter<EventEnvelope> emitter;

    private final RouteTable routeTable;
    private final String defaultTopic;

    public RoutingReflector(RouteTable routeTable) {
        this.routeTable = routeTable;

        var configProvider = ConfigProvider.getConfig();
        defaultTopic = configProvider.getValue("defaultTopic", String.class);
        
        logger.info("default topic: " + defaultTopic);
    }

    @Override
    public void reflect(Message<JsonObject> message, EventEnvelope event) {
        var topics = routeTable.getTopics(event);

        if (topics.isEmpty()) {
            emit(defaultTopic, event);
            return;
        }

        for (var topic: topics) {
            emit(topic, event);
        }
    }

    private void emit(String topic, EventEnvelope event) {
        OutgoingKafkaRecordMetadata<?> metadata = OutgoingKafkaRecordMetadata.builder()
                .withTopic(topic)
                .build();

        emitter.send(Message.of(event).addMetadata(metadata));
    }
}

The reflector channel is initialized in the application.properties resource:

mp.messaging.outgoing.reflector.cloud-events-source
    = reflectorservice

mp.messaging.outgoing.reflector.connector
    = smallrye-kafka

mp.messaging.outgoing.reflector.value.serializer
    = org.apache.kafka.common.serialization.StringSerializer

mp.messaging.outgoing.reflector.cloud-events-mode
    = structured

mp.messaging.outgoing.reflector.cloud-events-type
    = reflector

The emit function appends the metadata and sends the message via the emitter.