CDI Messaging Example

@Inject
@Channel("my-outgoing-topic")
private Emitter<String> emitter;

@Incoming("my-ingoing-topic")
public void consume(final String message) {
  emitter.send(process(message));
  emitter.complete();
}

@Inject
private Event<Out> sender;

public void onMessage(@Observes final In message) {
    sender.fire(process(message));
}

public record Out( (1)
        Integer partition, Long timestamp, (2)
        Map<String, byte[]> headers, Object key, Object value, (3)
        CompletableFuture<Void> completion) { (4)
}

@ApplicationScoped
public class MyKafkaProducer {
    private Producer<JsonValue, JsonValue> producer;
    private String topic;

    @PostConstruct
    private void init() { (1)
        producer = createKafkaProducer(); // standard creation
        topic = getTopicName(); // from conf generally
    }

    public void onMessage(@Observes final Out message) { (2)
        final var record = new ProducerRecord<>(
            topic, message.partition(), message.timestamp(),
            message.key(), message.value());
        if (message.headers() != null) {
            message.headers().forEach(record.headers()::add);
        }
        if (message.completion() == null) { (3)
            producer.send(record);
        } else {
            producer.send(record, new NotifyOnCompletion(message.completion()));
        }
    }

    private record NotifyOnCompletion(CompletableFuture<OutgoingMessage.Completion> promise) implements Callback {
        @Override
        public void onCompletion(final RecordMetadata metadata, final Exception exception) {
            if (exception  != null) {
                promise.completeExceptionally(exception);
            } else {
                promise.complete(new OutgoingMessage.Completion(metadata.partition(), metadata.offset(), metadata.timestamp()));
            }
        }
    }
}

@Log
@ApplicationScoped
public class MyKafkaConsumer {
    private Consumer<?, ?> consumer;
    private AtomicBoolean running;
    private ExecutorService pool;
    private ExecutorService appPool;

    public void forceStart(@Observes @Intialized(ApplicationScoped.class) final Object start) { (1)
        // force the instance to start when the app begins
    }

    @PostConstruct
    private void onConstruct() {
        (2)
        final var observerMethods = beanManager.resolveObserverMethods(new In(0, 0, 0, null, null, null));
        if (observerMethods.isEmpty()) {
            return;
        }

        (3)
        consumer = newKafkaConsumer();
        consumer.subscribe(List.of(getTopic()));

        (4)
        final var pool = Executors.newFixedThreadPool(Math.max(1, getPollerThreadNumber()), new ThreadFactory() {
            private final AtomicInteger counter = new AtomicInteger();

            @Override
            public Thread newThread(final Runnable r) {
                return new Thread(r, "kafka-" + conf.topic() + "-consumer-poll-" + counter.incrementAndGet());
            }
        });
        final var appPool = Executors.newFixedThreadPool(Math.max(1, getProcessorThreadNumber()), new ThreadFactory() {
            private final AtomicInteger counter = new AtomicInteger();

            @Override
            public Thread newThread(final Runnable r) {
                return new Thread(r, "kafka-" + conf.topic() + "-consumer-process-" + counter.incrementAndGet());
            }
        });

        final java.util.function.Consumer<In> invoke = i -> { (5)
            for (final var m: observerMethods) {
                m.notify(i);
            }
        };

        (6)
        final var running = new AtomicBoolean(true);
        final var pollTimeout = Duration.ofMillis(100);
        while (running.get()) {
            try {
                final var records = consumer.poll(pollTimeout);
                for (final var record: records) {
                    appPool.execute(() -> invoke.accept(new In( (7)
                        record.partition(),
                        record.offset(),
                        record.timestamp(),
                        StreamSupport.stream(record.headers().spliterator(), false)
                            .collect(toMap(Header::key, Header::value, (a, b) -> a)),
                        record.key(),
                        record.value())));
                }
            } catch (final RuntimeException re) {
                log.log(SEVERE, re, re::getMessage);
            }
        }

        return consumer;
    }

    @PreDestroy
    private void onDestroy() { (8)
        running.set(false);
        consumer.close();
        pool.shutdown();
        appPool.shutdown();
        try {
            if (!pool.awaitTermination(1, TimeUnit.MINUTES) || !appPool.awaitTermination(1, TimeUnit.MINUTES)) {
                log.warning(() -> "Can't stop " + conf.topic() + " consumer in 1mn, giving up");
            }
        } catch (final InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}