Нам нужно перехватывать все запросы в Kafka и отправлять в систему аудита или телеметрии. Как бы ты это сделал?

Для перехвата всех запросов в Kafka и отправки их в систему аудита или телеметрии можно использовать несколько подходов:

1. Использование Kafka Interceptor API — Kafka предоставляет интерфейсы ProducerInterceptor и ConsumerInterceptor, которые позволяют перехватывать сообщения при отправке и приёме. В этих перехватчиках можно реализовать логику отправки данных в систему аудита.

public class AuditProducerInterceptor implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // Отправляем данные в систему аудита
        auditService.send(record);
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
    @Override
    public void close() {}
    @Override
    public void configure(Map<String, ?> configs) {}
}

2. Прокси-сервис — создать промежуточный сервис, который принимает сообщения от продюсеров, отправляет их в Kafka и одновременно дублирует в систему аудита.

3. Использование Kafka Connect с Sink Connector — настроить коннектор, который будет читать из топика и отправлять данные в систему аудита.

4. Обработка на стороне потребителей — если аудит важен для потребления, можно добавить логику отправки в аудит в коде консьюмера.

Наиболее прозрачным и масштабируемым вариантом для Java-приложений является использование ProducerInterceptor и ConsumerInterceptor, так как они встроены в Kafka-клиент и не требуют изменения архитектуры.