Параметры подтверждения сообщения Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "0");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 
Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "key", "value"));
// Никакого ожидания! Сразу переходим к следующей операции

props.put("acks", "1");
// ...
Future<RecordMetadata> future = producer.send(new ProducerRecord<>("my-topic", "key", "value"));
// Можем ждать результат, если хотим быть уверены в доставке
RecordMetadata metadata = future.get();

props.put("acks", "all");
// Также полезно установить хорошее значение для повторных попыток
props.put("retries", Integer.MAX_VALUE);
props.put("delivery.timeout.ms", 120000); // 2 минуты максимум на доставку

// На стороне брокера:
min.insync.replicas=2
 
// На стороне клиента обрабатываем возможные исключения:
try {
    producer.send(record).get();
} catch (NotEnoughReplicasException e) {
    // Логика обработки ситуации, когда недостаточно реплик
    // Например, приостановка отправки и повторная попытка позже
}

// Более сложный пример с обработкой результатов
producer.send(record, (recordMetadata, exception) -> {
    if (exception != null) {
        // Обработка исключения при доставке
        log.error("Ошибка отправки сообщения: {}", exception.getMessage());
        // Возможно, повторная отправка или сохранение в локальный буфер
    } else {
        // Сообщение доставлено
        log.debug("Сообщение отправлено в топик {} партицию {} с offset {}", 
                  recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
    }
});

// Оптимизация для высокой пропускной способности при acks=all
props.put("acks", "all");
props.put("batch.size", 64000);  // Разрешаем более крупные пакеты
props.put("linger.ms", 5);       // Позволяем немного подождать для формирования пакета
props.put("compression.type", "snappy");  // Экономим сетевой трафик через компрессию

// На стороне брокера:
replica.lag.time.max.ms=10000  // 10 секунд максимального отставания

Properties properties = new Properties();
properties.put("bootstrap.servers", "kafka-broker1:9092,kafka-broker2:9092");
properties.put("acks", "0");  // Максимальная производительность
properties.put("compression.type", "lz4");
properties.put("batch.size", 131072);  // 128KB пакеты для эффективности
properties.put("linger.ms", 20);  // Собираем больше сообщений в пакет

Properties properties = new Properties();
properties.put("bootstrap.servers", "финансовый-кластер-1:9092,финансовый-кластер-2:9092");
properties.put("acks", "all");  // Максимальная надёжность
properties.put("min.insync.replicas", "2");  // Минимум две реплики должны подтвердить запись
properties.put("enable.idempotence", true);  // Избегаем дублирования сообщений
properties.put("retries", Integer.MAX_VALUE);
properties.put("max.in.flight.requests.per.connection", 5);
properties.put("delivery.timeout.ms", 30000);  // 30 секунд на доставку максимум

Producer<String, String> producer = new KafkaProducer<>(properties);
producer.initTransactions();
 
try {
    producer.beginTransaction();
    // Отправляем связанные сообщения, которые должны быть обработаны атомарно
    producer.send(new ProducerRecord<>("payments", "client1", "debit:100"));
    producer.send(new ProducerRecord<>("payments", "client2", "credit:100"));
    // Фиксируем транзакцию только если все сообщения успешно отправлены
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
    throw e;
}

// Конфигурация MirrorMaker для кросс-региональной репликации
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "datacenter-1:9092");
consumerProps.put("group.id", "mirror-maker-group");
consumerProps.put("auto.offset.reset", "earliest");
 
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "datacenter-2:9092");
producerProps.put("acks", "all");
producerProps.put("retries", "3");

// Для регулярной телеметрии
Properties regularTelemetryProps = new Properties();
regularTelemetryProps.put("bootstrap.servers", "iot-cluster:9092");
regularTelemetryProps.put("acks", "1");  // Компромисс
regularTelemetryProps.put("compression.type", "snappy");
regularTelemetryProps.put("batch.size", 65536);
Producer<String, String> regularProducer = new KafkaProducer<>(regularTelemetryProps);
 
// Для критических событий
Properties criticalEventsProps = new Properties();
criticalEventsProps.put("bootstrap.servers", "iot-cluster:9092");
criticalEventsProps.put("acks", "all");  // Максимальная надежность
criticalEventsProps.put("enable.idempotence", true);
criticalEventsProps.put("retries", Integer.MAX_VALUE);
Producer<String, String> criticalProducer = new KafkaProducer<>(criticalEventsProps);

// Пример логики на стороне IoT-шлюза
class IoTGateway {
    private Queue<SensorReading> criticalBuffer = new ConcurrentLinkedQueue<>();
    
    public void processSensorReading(SensorReading reading) {
        if (reading.isCritical()) {
            // Буферизуем критичные чтения, пока не получим подтверждение
            criticalBuffer.add(reading);
            sendCriticalReading(reading);
        } else {
            // Обычные показания отправляем обычным способом
            sendRegularReading(reading);
        }
    }
    
    private void sendCriticalReading(SensorReading reading) {
        criticalProducer.send(
            new ProducerRecord<>("critical-events", reading.toJson()),
            (metadata, exception) -> {
                if (exception == null) {
                    // Удаляем из буфера только после подтверждения
                    criticalBuffer.remove(reading);
                } else {
                    // Повторная отправка будет автоматической через настройку retries
                    log.error("Ошибка отправки критического события: {}", exception.getMessage());
                }
            }
        );
    }
}

// Настройка на уровне брокера
min.insync.replicas=2
 
// Или на уровне топика
kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic critical-data --config min.insync.replicas=2

// Пример конфигурации для ситуаций, где потеря данных абсолютно недопустима
Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
props.put("acks", "all");
props.put("enable.idempotence", true);
// Обработка ошибок, связанных с недостаточным количеством реплик
props.put("retries", Integer.MAX_VALUE);
// Не более 5 одновременных запросов для сохранения порядка сообщений
props.put("max.in.flight.requests.per.connection", 5);

// Более агрессивная буферизация для acks=all
props.put("batch.size", 131072);  // 128KB
props.put("linger.ms", 10);       // Ожидание до 10 мс для формирования пакета
props.put("buffer.memory", 67108864); // 64MB буфер в памяти

props.put("compression.type", "zstd");  // Zstandard дает отличный баланс между степенью и скоростью

props.put("retries", Integer.MAX_VALUE);  // Неограниченные попытки
props.put("retry.backoff.ms", 150);       // 150 мс между попытками
props.put("delivery.timeout.ms", 120000);  // Общий таймаут = 2 минуты

// Для некритичных данных
Properties fastProps = new Properties();
fastProps.put("acks", "1");
Producer<String, String> fastProducer = new KafkaProducer<>(fastProps);
 
// Для критичных данных
Properties reliableProps = new Properties();
reliableProps.put("acks", "all");
reliableProps.put("min.insync.replicas", "2");
Producer<String, String> reliableProducer = new KafkaProducer<>(reliableProps);
 
// Использование в коде
if (message.isCritical()) {
    reliableProducer.send(new ProducerRecord<>("critical-topic", message.key(), message.value()));
} else {
    fastProducer.send(new ProducerRecord<>("regular-topic", message.key(), message.value()));
}

// Включение метрик JMX в продюсере
props.put("metrics.recording.level", "INFO");
props.put("metrics.sample.window.ms", 30000);

// Псевдокод для адаптивной настройки параметра acks
void adjustProducerConfig(Producer producer, KafkaMetrics metrics) {
    if (metrics.getClusterLoad() > HIGH_LOAD_THRESHOLD) {
        // При высокой нагрузке снижаем требования к надежности
        properties.put("acks", "1");
        producer.updateConfig(properties);
    } else {
        // При нормальной нагрузке максимальная надежность
        properties.put("acks", "all");
        producer.updateConfig(properties);
    }
}

// На уровне брокера
unclean.leader.election.enable=false
 
// Или на уровне топика
kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic financial-transactions --config unclean.leader.election.enable=false

// Настройка брокера
replica.lag.time.max.ms=10000  // 10 секунд максимального отставания

// Более реалистичное значение для высоконагруженных систем
replica.lag.time.max.ms=30000  // 30 секунд максимального отставания

// Увеличение максимального размера запроса репликации
replica.fetch.max.bytes=10485760  // 10MB (вместо дефолтного 1MB)