Партиционирование Kafka и доставка сообщений из нескольких партиций

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
 
// Партиционирование по ключу - все сообщения одного пользователя в одну партицию
for (int i = 0; i < 100; i++) {
    ProducerRecord<String, String> record = new ProducerRecord<>(
        "orders", 
        "user_" + (i % 10),  // Десять разных пользователей
        "order_data_" + i
    );
    producer.send(record);
}
 
// Round-robin - ключ null, распределение по всем партициям
for (int i = 0; i < 100; i++) {
    ProducerRecord<String, String> record = new ProducerRecord<>(
        "logs",
        null,  // Без ключа
        "log_entry_" + i
    );
    producer.send(record);
}
 
producer.close();

public class VipPartitioner implements Partitioner {
    
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                        Object value, byte[] valueBytes, Cluster cluster) {
        
        int numPartitions = cluster.partitionCountForTopic(topic);
        
        if (key != null && key.toString().startsWith("VIP_")) {
            // VIP клиенты всегда в партицию 0
            return 0;
        }
        
        // Остальные равномерно по партициям 1..N
        int hash = Math.abs(key.hashCode());
        return (hash % (numPartitions - 1)) + 1;
    }
    
    @Override
    public void close() {}
    
    @Override
    public void configure(Map<String, ?> configs) {}
}

public class TimeWindowPartitioner implements Partitioner {
    
    private static final long WINDOW_MS = 60_000; // 1 минута
    
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                        Object value, byte[] valueBytes, Cluster cluster) {
        
        int numPartitions = cluster.partitionCountForTopic(topic);
        
        // Округляем timestamp до минуты и хешируем
        long currentWindow = System.currentTimeMillis() / WINDOW_MS;
        int hash = Math.abs((int) currentWindow);
        
        return hash % numPartitions;
    }
    
    @Override
    public void close() {}
    
    @Override 
    public void configure(Map<String, ?> configs) {}
}

public class IdempotentOrderProcessor {
    
    private final Set<String> processedOrderIds = ConcurrentHashMap.newKeySet();
    
    public void processOrder(ConsumerRecord<String, Order> record) {
        Order order = record.value();
        String orderId = order.getId();
        
        // Проверяем, обрабатывали ли уже этот заказ
        if (processedOrderIds.contains(orderId)) {
            System.out.println("Заказ " + orderId + " уже обработан, пропускаем");
            return;
        }
        
        try {
            // Реальная бизнес-логика
            database.insertOrder(order);
            paymentService.charge(order.getAmount());
            
            // Помечаем как обработанный только после успеха
            processedOrderIds.add(orderId);
            
        } catch (Exception e) {
            // Ошибка - не помечаем обработанным, повторим при следующей попытке
            throw e;
        }
    }
}

public class OrderedMultithreadConsumer {
    
    private final ExecutorService executor = Executors.newFixedThreadPool(4);
    private final Map<String, BlockingQueue<ConsumerRecord<String, Event>>> userQueues = 
        new ConcurrentHashMap<>();
    
    public void consume() {
        while (true) {
            ConsumerRecords<String, Event> records = consumer.poll(Duration.ofMillis(100));
            
            for (ConsumerRecord<String, Event> record : records) {
                String userId = record.key();
                
                // Каждый пользователь получает свою очередь
                BlockingQueue<ConsumerRecord<String, Event>> userQueue = 
                    userQueues.computeIfAbsent(userId, k -> new LinkedBlockingQueue<>());
                
                userQueue.offer(record);
                
                // Запускаем обработчик, если его ещё нет
                executor.submit(() -> processUserQueue(userId, userQueue));
            }
        }
    }
    
    private void processUserQueue(String userId, BlockingQueue<ConsumerRecord<String, Event>> queue) {
        try {
            ConsumerRecord<String, Event> record = queue.poll(1, TimeUnit.SECONDS);
            if (record != null) {
                processEvent(record.value());
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    
    private void processEvent(Event event) {
        // Тяжелая бизнес-логика
    }
}

public class OutOfOrderEventProcessor {
    
    private final NavigableMap<Long, List<Event>> eventBuffer = new TreeMap<>();
    private final long maxOutOfOrderMillis = 30_000; // 30 секунд
    private long currentWatermark = 0;
    
    public void processEvent(Event event) {
        long eventTime = event.getTimestamp();
        
        // Событие слишком старое, пропущен watermark
        if (eventTime < currentWatermark) {
            metrics.incrementLateEvent();
            handleLateEvent(event);
            return;
        }
        
        // Буферизуем событие
        eventBuffer.computeIfAbsent(eventTime, k -> new ArrayList<>()).add(event);
        
        // Продвигаем watermark если накопилось достаточно событий
        long newWatermark = eventTime - maxOutOfOrderMillis;
        if (newWatermark > currentWatermark) {
            processBufferedEvents(currentWatermark, newWatermark);
            currentWatermark = newWatermark;
        }
    }
    
    private void processBufferedEvents(long fromTime, long toTime) {
        NavigableMap<Long, List<Event>> toProcess = 
            eventBuffer.subMap(fromTime, true, toTime, true);
        
        // Обрабатываем в порядке возрастания timestamp
        for (Map.Entry<Long, List<Event>> entry : toProcess.entrySet()) {
            for (Event event : entry.getValue()) {
                handleEvent(event);
            }
        }
        
        // Очищаем обработанные
        toProcess.clear();
    }
    
    private void handleLateEvent(Event event) {
        // Либо игнорируем, либо кладём в side output, либо переобрабатываем
        lateEventsQueue.offer(event);
    }
    
    private void handleEvent(Event event) {
        // Основная бизнес-логика
    }
}

Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 
// Настройки надёжности
producerProps.put("acks", "all"); // Ждём все ISR реплики
producerProps.put("retries", 3); // Три попытки при ошибке
producerProps.put("max.in.flight.requests.per.connection", 5); // Параллельные запросы
producerProps.put("enable.idempotence", true); // Защита от дубликатов
 
// Настройки производительности
producerProps.put("batch.size", 32768); // 32KB батч
producerProps.put("linger.ms", 10); // Ждём 10мс заполнения батча
producerProps.put("compression.type", "lz4"); // Быстрая компрессия
producerProps.put("buffer.memory", 67108864); // 64MB буфер
 
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
consumerProps.put("group.id", "payment-processor");
consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 
// Контроль offset'ов
consumerProps.put("enable.auto.commit", false); // Ручной коммит
consumerProps.put("auto.offset.reset", "earliest"); // При отсутствии offset'а - с начала
 
// Настройки fetch
consumerProps.put("fetch.min.bytes", 1024); // Минимум 1KB
consumerProps.put("fetch.max.wait.ms", 500); // Ждём max 500мс
consumerProps.put("max.poll.records", 500); // До 500 записей за раз
 
// Heartbeat и сессия
consumerProps.put("heartbeat.interval.ms", 3000);
consumerProps.put("session.timeout.ms", 30000);
consumerProps.put("max.poll.interval.ms", 300000); // 5 минут на обработку
 
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("payments"));
 
try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        
        for (ConsumerRecord<String, String> record : records) {
            try {
                processPayment(record.value());
            } catch (Exception e) {
                handleProcessingError(record, e);
            }
        }
        
        // Коммитим только после успешной обработки всего батча
        consumer.commitSync();
    }
} catch (WakeupException e) {
    // Корректное завершение
} finally {
    consumer.close();
}

// Пример мониторинга состояния реплик через AdminClient
Properties adminProps = new Properties();
adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092");
 
AdminClient adminClient = AdminClient.create(adminProps);
 
try {
    // Получаем описание топика
    DescribeTopicsResult result = adminClient.describeTopics(
        Collections.singleton("critical-events")
    );
    
    TopicDescription description = result.values().get("critical-events").get();
    
    for (TopicPartitionInfo partition : description.partitions()) {
        System.out.println("Партиция " + partition.partition() + ":");
        System.out.println("  Лидер: " + partition.leader().id());
        
        // Список всех реплик
        System.out.print("  Реплики: ");
        partition.replicas().forEach(node -> System.out.print(node.id() + " "));
        System.out.println();
        
        // In-sync реплики
        System.out.print("  ISR: ");
        partition.isr().forEach(node -> System.out.print(node.id() + " "));
        System.out.println();
        
        // Проверяем отставание
        if (partition.isr().size() < partition.replicas().size()) {
            System.out.println("  ВНИМАНИЕ: Не все реплики в ISR!");
        }
    }
    
} catch (Exception e) {
    e.printStackTrace();
} finally {
    adminClient.close();
}

public class LagMonitor {
  
  private final AdminClient adminClient;
  private final String groupId;
  
  public Map<TopicPartition, Long> calculateLag() throws Exception {
      Map<TopicPartition, Long> lags = new HashMap<>();
      
      // Получаем offset'ы консьюмер группы
      ListConsumerGroupOffsetsResult offsetsResult = 
          adminClient.listConsumerGroupOffsets(groupId);
      Map<TopicPartition, OffsetAndMetadata> consumerOffsets = 
          offsetsResult.partitionsToOffsetAndMetadata().get();
      
      // Получаем end offset'ы партиций
      Map<TopicPartition, Long> endOffsets = adminClient.listOffsets(
          consumerOffsets.keySet().stream()
              .collect(Collectors.toMap(
                  tp -> tp,
                  tp -> OffsetSpec.latest()
              ))
      ).all().get().entrySet().stream()
          .collect(Collectors.toMap(
              Map.Entry::getKey,
              e -> e.getValue().offset()
          ));
      
      // Вычисляем лаг для каждой партиции
      for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : consumerOffsets.entrySet()) {
          TopicPartition tp = entry.getKey();
          long consumerOffset = entry.getValue().offset();
          long endOffset = endOffsets.getOrDefault(tp, 0L);
          long lag = endOffset - consumerOffset;
          
          lags.put(tp, lag);
          
          if (lag > 100000) {
              System.out.printf("ВНИМАНИЕ: Партиция %s отстаёт на %d сообщений%n",
                  tp, lag);
          }
      }
      
      return lags;
  }
}

public class MultiStrategyProducer implements AutoCloseable {
    
    private final KafkaProducer<String, String> producer;
    private final String topic;
    private final boolean useKeys;
    private final AtomicLong sentCount = new AtomicLong(0);
    private volatile boolean running = true;
    
    public MultiStrategyProducer(Properties props, String topic, boolean useKeys) {
        // Базовая конфигурация для надёжности
        props.put("acks", "all");
        props.put("retries", 3);
        props.put("enable.idempotence", true);
        props.put("compression.type", "lz4");
        
        this.producer = new KafkaProducer<>(props);
        this.topic = topic;
        this.useKeys = useKeys;
    }
    
    public void sendEvents(int count, String eventType) {
        Random random = new Random();
        
        for (int i = 0; i < count && running; i++) {
            String userId = "user_" + random.nextInt(1000);
            String eventData = String.format(
                "{\"type\":\"%s\",\"userId\":\"%s\",\"timestamp\":%d,\"sessionId\":\"%s\"}",
                eventType, userId, System.currentTimeMillis(), UUID.randomUUID()
            );
            
            // Ключ или null в зависимости от стратегии
            String key = useKeys ? userId : null;
            
            ProducerRecord<String, String> record = 
                new ProducerRecord<>(topic, key, eventData);
            
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.err.printf("Ошибка отправки: %s%n", exception.getMessage());
                } else {
                    sentCount.incrementAndGet();
                    if (sentCount.get() % 1000 == 0) {
                        System.out.printf("Отправлено %d событий в партицию %d%n",
                            sentCount.get(), metadata.partition());
                    }
                }
            });
            
            // Небольшая задержка для реалистичности
            try {
                Thread.sleep(random.nextInt(5));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
    
    @Override
    public void close() {
        running = false;
        producer.flush();
        producer.close(Duration.ofSeconds(10));
        System.out.printf("Продюсер закрыт. Всего отправлено: %d%n", sentCount.get());
    }
}

public class PartitionedConsumer implements Runnable {
    
    private final KafkaConsumer<String, String> consumer;
    private final String topic;
    private final int processingDelayMs;
    private final AtomicLong processedCount = new AtomicLong(0);
    private volatile boolean running = true;
    
    public PartitionedConsumer(Properties props, String topic, int processingDelayMs) {
        props.put("enable.auto.commit", "false");
        props.put("max.poll.records", 500);
        props.put("session.timeout.ms", 30000);
        props.put("max.poll.interval.ms", 300000);
        
        this.consumer = new KafkaConsumer<>(props);
        this.topic = topic;
        this.processingDelayMs = processingDelayMs;
    }
    
    @Override
    public void run() {
        try {
            consumer.subscribe(Collections.singleton(topic));
            
            while (running) {
                ConsumerRecords<String, String> records = 
                    consumer.poll(Duration.ofMillis(100));
                
                if (records.isEmpty()) {
                    continue;
                }
                
                // Группируем по партициям для статистики
                Map<Integer, Integer> partitionCounts = new HashMap<>();
                
                for (ConsumerRecord<String, String> record : records) {
                    try {
                        processEvent(record);
                        partitionCounts.merge(record.partition(), 1, Integer::sum);
                    } catch (Exception e) {
                        System.err.printf("Ошибка обработки события из партиции %d: %s%n",
                            record.partition(), e.getMessage());
                    }
                }
                
                // Коммитим только после успешной обработки всего батча
                consumer.commitSync();
                
                // Логируем прогресс
                long total = processedCount.addAndGet(records.count());
                if (total % 5000 == 0) {
                    System.out.printf("Обработано %d событий. Распределение: %s%n",
                        total, partitionCounts);
                }
            }
            
        } catch (WakeupException e) {
            // Нормальное завершение через shutdown hook
        } finally {
            consumer.close();
            System.out.printf("Консьюмер остановлен. Обработано: %d%n", processedCount.get());
        }
    }
    
    private void processEvent(ConsumerRecord<String, String> record) throws Exception {
        // Симуляция реальной обработки
        if (processingDelayMs > 0) {
            Thread.sleep(processingDelayMs);
        }
        
        // Здесь была бы реальная бизнес-логика
        // Запись в БД, обогащение данных, агрегация и т.д.
    }
    
    public void shutdown() {
        running = false;
        consumer.wakeup();
    }
}

public class EventProcessorApp {
    
    public static void main(String[] args) throws Exception {
        String bootstrapServers = "localhost:9092";
        String topic = "user-events";
        
        // Настройки продюсера
        Properties producerProps = new Properties();
        producerProps.put("bootstrap.servers", bootstrapServers);
        producerProps.put("key.serializer", StringSerializer.class.getName());
        producerProps.put("value.serializer", StringSerializer.class.getName());
        
        // Запускаем два продюсера - один с ключами, один без
        MultiStrategyProducer keyedProducer = 
            new MultiStrategyProducer(producerProps, topic, true);
        MultiStrategyProducer roundRobinProducer = 
            new MultiStrategyProducer(producerProps, topic, false);
        
        // Настройки консьюмера
        Properties consumerProps = new Properties();
        consumerProps.put("bootstrap.servers", bootstrapServers);
        consumerProps.put("group.id", "event-processors");
        consumerProps.put("key.deserializer", StringDeserializer.class.getName());
        consumerProps.put("value.deserializer", StringDeserializer.class.getName());
        
        // Запускаем три консьюмера в разных потоках
        List<PartitionedConsumer> consumers = new ArrayList<>();
        List<Thread> threads = new ArrayList<>();
        
        for (int i = 0; i < 3; i++) {
            Properties props = new Properties(consumerProps);
            PartitionedConsumer consumer = 
                new PartitionedConsumer(props, topic, 2); // 2мс задержка
            consumers.add(consumer);
            
            Thread thread = new Thread(consumer, "Consumer-" + i);
            threads.add(thread);
            thread.start();
        }
        
        // Graceful shutdown
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("Завершение работы...");
            consumers.forEach(PartitionedConsumer::shutdown);
            keyedProducer.close();
            roundRobinProducer.close();
        }));
        
        // Генерируем нагрузку
        new Thread(() -> keyedProducer.sendEvents(100000, "click")).start();
        new Thread(() -> roundRobinProducer.sendEvents(50000, "login")).start();
        
        // Ждём завершения
        for (Thread thread : threads) {
            thread.join();
        }
    }
}

public class ResilientEventProcessor {
  
  private final KafkaProducer<String, String> dlqProducer;
  private final String dlqTopic;
  private final int maxRetries;
  
  public void processWithRetry(ConsumerRecord<String, String> record) {
      int attempt = 0;
      Exception lastException = null;
      
      while (attempt < maxRetries) {
          try {
              // Основная логика обработки
              processEventLogic(record.value());
              return; // Успех, выходим
              
          } catch (TransientException e) {
              // Временная ошибка - retry с backoff
              lastException = e;
              attempt++;
              
              if (attempt < maxRetries) {
                  long delayMs = (long) Math.pow(2, attempt) * 100;
                  System.out.printf("Retry %d/%d после %dмс для события из партиции %d%n",
                      attempt, maxRetries, delayMs, record.partition());
                  
                  try {
                      Thread.sleep(delayMs);
                  } catch (InterruptedException ie) {
                      Thread.currentThread().interrupt();
                      break;
                  }
              }
              
          } catch (PermanentException e) {
              // Необратимая ошибка - сразу в DLQ
              System.err.printf("Необратимая ошибка, событие отправлено в DLQ: %s%n",
                  e.getMessage());
              sendToDLQ(record, e);
              return;
          }
      }
      
      // Исчерпаны попытки - в DLQ
      if (lastException != null) {
          System.err.printf("Исчерпаны %d попыток, событие в DLQ%n", maxRetries);
          sendToDLQ(record, lastException);
      }
  }
  
  private void sendToDLQ(ConsumerRecord<String, String> original, Exception error) {
      String dlqValue = String.format(
          "{\"original\":%s,\"error\":\"%s\",\"timestamp\":%d}",
          original.value(),
          error.getMessage().replace("\"", "\\\""),
          System.currentTimeMillis()
      );
      
      ProducerRecord<String, String> dlqRecord = 
          new ProducerRecord<>(dlqTopic, original.key(), dlqValue);
      
      dlqProducer.send(dlqRecord, (metadata, exception) -> {
          if (exception != null) {
              // Если даже DLQ недоступен - пишем в файл
              logToFile(original, error);
          }
      });
  }
  
  private void logToFile(ConsumerRecord<String, String> record, Exception error) {
      // Fallback на случай полного отказа Kafka
      try (FileWriter fw = new FileWriter("failed_events.log", true)) {
          fw.write(String.format("%s|%s|%s%n",
              System.currentTimeMillis(), record.value(), error.getMessage()));
      } catch (IOException e) {
          // Ну тут уже совсем всё плохо
          e.printStackTrace();
      }
  }
}