Почему Kafka является безопасной системой хранения данных?

Почему Kafka является безопасной системой хранения данных?

Kafka НЕ просто хранилище, а распределённая система очереди сообщений. Её безопасность основана на нескольких механизмах.

Kafka как система хранения данных

Kafka отличается от БД тем, что это event stream, а не CRUD хранилище. Но она хранит данные безопасно благодаря:

1. Replication (Репликация)

Каждое сообщение в Kafka репликируется на несколько broker'ов:

Partition: [msg1] [msg2] [msg3] [msg4]
              |       |      |      |
           Leader  Broker1 Broker2 Broker3
           (запись)

Репликация гарантирует:

• ✅ Отказоустойчивость — если 1 broker упадёт, данные не потеряны
• ✅ Дублирование — каждое сообщение скопировано (по умолчанию 3 копии)
• ✅ Долговечность — данные сохранены на диск

Конфигурация:

log.replication.factor=3  # 3 копии каждого сообщения
min.insync.replicas=2     # Хотя бы 2 должны подтвердить запись

2. Persistence (Постоянное хранилище)

Kafka пишет все сообщения на диск:

Producer → Memory Buffer → Disk (fsync) → Consumer
                            ↓
                      Permanent Log

Конфигурация:

log.flush.interval.messages=10000  # Писать на диск каждые 10к сообщений
log.flush.interval.ms=1000          # Или каждую 1 сек

Безопасность:

• ❌ Если сервер упадёт БЕЗ fsync → данные теряются
• ✅ Если сервер упадёт ПОСЛЕ fsync → данные сохранены

3. Acks конфигурация (гарантии доставки)

Producer отправляет с разными уровнями гарантий:

Примеры:

// Небезопасно (acks=0)
Properties props = new Properties();
props.put("acks", "0");
props.put("retries", 0);
Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("topic", "msg"));
// Сообщение может потеряться!

// Безопасно (acks=all)
props.put("acks", "all");
props.put("retries", 3);
props.put("min.insync.replicas", 2);
Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("topic", "msg"),
    (record, ex) -> {
        if (ex != null) {
            System.out.println("Failed to send: " + ex);
        }
    });

4. Consumer Offsets и Exactly-Once Semantics

Kafka отслеживает, какие сообщения обработаны:

Producer → [msg1] [msg2] [msg3] [msg4]
             ↑                    ↑
          Offset=0            Offset=3
             
Consumer → читает msg1 → offset становится 1
        → читает msg2 → offset становится 2

Гарантии:

1. At-Least-Once (обычно)

   consumer.subscribe(Arrays.asList("my-topic"));
   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
   for (ConsumerRecord<String, String> record : records) {
       process(record);
       consumer.commitSync(); // Сохранить offset
   }
   

   • ✅ Каждое сообщение обработано минимум 1 раз
   • ⚠️ Может быть обработано 2+ раза (если crash)

2. Exactly-Once (с Kafka transactions)

   producer.beginTransaction();
   try {
       producer.send(new ProducerRecord<>("topic", msg));
       producer.commitTransaction();
   } catch (Exception e) {
       producer.abortTransaction();
   }
   

   • ✅ Каждое сообщение обработано ровно 1 раз
   • ⚠️ Медленнее, требует координации

5. Log Compaction (для data retention)

Kafka хранит все сообщения, но может очищать старые:

Original:  [key1:v1] [key1:v2] [key1:v3] [key2:v1] [key2:v2]

After compaction: [key1:v3] [key2:v2]
(Сохранены только последние значения)

Конфигурация:

log.cleanup.policy=compact
log.segment.ms=86400000  # Сегмент = 1 день

6. Аутентификация и авторизация

SASL/SSL для безопасного доступа:

# Broker
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="user" password="pass";
ssl.truststore.location=/path/to/truststore
ssl.truststore.password=password

// Producer
props.put("security.protocol", "SASL_SSL");
props.put("sasl.mechanism", "SCRAM-SHA-256");
props.put("sasl.jaas.config", 
    "org.apache.kafka.common.security.scram.ScramLoginModule required " +
    "username=\"user\" password=\"pass\";");

Сравнение: Kafka vs БД

Реальный пример: Payment Service

Небезопасная реализация:

// Без гарантий! Если crash → потеряем платёж
producer.send(new ProducerRecord<>("payments", payment.toString()));
processPayment(payment);

Безопасная реализация:

Properties props = new Properties();
props.put("acks", "all");
props.put("retries", 3);
props.put("enable.idempotence", true); // Не дублировать при retry
props.put("min.insync.replicas", 2);

Producer<String, String> producer = new KafkaProducer<>(props);

Future<RecordMetadata> future = producer.send(
    new ProducerRecord<>("payments", String.valueOf(payment.getId()), 
                         serializePayment(payment)),
    (record, ex) -> {
        if (ex != null) {
            logger.error("Failed to send payment: ", ex);
            // Retry логика
        } else {
            logger.info("Payment sent: " + record.offset());
        }
    }
);

// Дождаться подтверждения ВСЕ реплик
RecordMetadata metadata = future.get(10, TimeUnit.SECONDS);
logger.info("Payment replicated to " + metadata.replicas().length + " brokers");

processPayment(payment);

Когда Kafka НЕ безопасна

❌ Ненадёжная конфигурация:

• acks=0 без подтверждений
• Нет репликации (replication.factor=1)
• Нет retry логики
• Отсутствует idempotence при resend

❌ Неправильное использование:

• Полагаться только на memory без fsync
• Не коммитить offsets
• Обрабатывать без дублирования (не At-Least-Once)

Вывод

Kafka ЯВЛЯЕТСЯ безопасной системой хранения благодаря:

1. ✅ Репликации на несколько broker'ов
2. ✅ Постоянному хранению на диск (fsync)
3. ✅ Гарантиям доставки (acks configuration)
4. ✅ Offset management для отслеживания прогресса
5. ✅ Transactions для exactly-once семантики
6. ✅ SASL/SSL для безопасности доступа

НО это требует правильной конфигурации. С неправильными настройками Kafka может потерять данные.