Позиционировани­е Kafka Consumer и Seek-операции

// Перемещение к конкретному смещению
void seek(TopicPartition partition, long offset);
 
// Перемещение к начальному смещению партиции
void seekToBeginning(Collection<TopicPartition> partitions);
 
// Перемещение к конечному смещению партиции
void seekToEnd(Collection<TopicPartition> partitions);
 
// Перемещение к смещению по временной метке
Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch);

// Пример перемещения к конкретному смещению
TopicPartition partition = new TopicPartition("my-topic", 0);
consumer.seek(partition, 1000L);

// Перемещение к началу всех назначенных партиций
consumer.seekToBeginning(consumer.assignment());
 
// Перемещение к концу конкретных партиций
Set<TopicPartition> partitionsToSeek = new HashSet<>();
partitionsToSeek.add(new TopicPartition("my-topic", 0));
partitionsToSeek.add(new TopicPartition("my-topic", 1));
consumer.seekToEnd(partitionsToSeek);

// Найти смещения для всех партиций заданного топика, соответствующие времени час назад
long timestamp = System.currentTimeMillis() - 3600000; // 1 час назад
List<PartitionInfo> partitionInfos = consumer.partitionsFor("my-topic");
Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
 
for (PartitionInfo info : partitionInfos) {
    timestampsToSearch.put(new TopicPartition("my-topic", info.partition()), timestamp);
}
 
Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(timestampsToSearch);
 
// Использование полученных смещений для позиционирования
for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsets.entrySet()) {
    if (entry.getValue() != null) {
        consumer.seek(entry.getKey(), entry.getValue().offset());
    }
}

// После seek необходимо зафиксировать новую позицию
consumer.seek(partition, newOffset);
// ... обработка сообщений ...
consumer.commitSync(); // Фиксирует текущие позиции для всех назначенных партиций

Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
offsetsToCommit.put(new TopicPartition("my-topic", 0), new OffsetAndMetadata(1000L));
consumer.commitSync(offsetsToCommit);

// Типичный паттерн использования seek
consumer.seek(partition, targetOffset);
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    // Обработка записей, вернувшихся из poll() ПОСЛЕ операции seek
    // ...
}

// Пример расширенного использования position и committed
Map<TopicPartition, Long> currentPositions = new HashMap<>();
Map<TopicPartition, OffsetAndMetadata> committedOffsets = new HashMap<>();
 
for (TopicPartition partition : consumer.assignment()) {
    currentPositions.put(partition, consumer.position(partition));
    OffsetAndMetadata committed = consumer.committed(partition);
    if (committed != null) {
        committedOffsets.put(partition, committed);
    }
}
 
// Анализ отставания
for (TopicPartition partition : currentPositions.keySet()) {
    long lag = currentPositions.get(partition);
    OffsetAndMetadata committed = committedOffsets.get(partition);
    if (committed != null) {
        lag -= committed.offset();
        System.out.printf("Партиция %s: необработанных сообщений: %d%n", 
                         partition, lag);
    }
}

// Отключение автоматического коммита
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
 
// Ручной коммит смещений
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // Обработка записи
        // ...
        
        // Фиксирование обработанной записи
        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
        offsets.put(new TopicPartition(record.topic(), record.partition()),
                   new OffsetAndMetadata(record.offset() + 1));
        consumer.commitSync(offsets);
    }
}

try {
    // Начало транзакции
    db.beginTransaction();
    
    // Обработка сообщений
    for (ConsumerRecord<String, String> record : records) {
        db.execute("INSERT INTO processed_messages VALUES (?)", record.value());
    }
    
    // Фиксирование транзакции
    db.commitTransaction();
    
    // Фиксирование смещений только после успешной обработки
    consumer.commitSync();
} catch (Exception e) {
    // Откат транзакции при ошибке
    db.rollbackTransaction();
    
    // При необходимости можно также откатить позицию чтения
    for (TopicPartition partition : consumer.assignment()) {
        OffsetAndMetadata committed = consumer.committed(partition);
        if (committed != null) {
            consumer.seek(partition, committed.offset());
        }
    }
}

// Хранение дополнительной информации вместе со смещением
String processingInfo = "{\"processor\":\"node-1\",\"timestamp\":\"" + 
                     System.currentTimeMillis() + "\"}";
OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(offset, processingInfo);
offsets.put(partition, offsetAndMetadata);
consumer.commitSync(offsets);

consumer.subscribe(Collections.singletonList("my-topic"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Сохранение позиций перед ребалансировкой
        for (TopicPartition partition : partitions) {
            savePosition(partition, consumer.position(partition));
        }
    }
 
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Восстановление позиций после ребалансировки
        for (TopicPartition partition : partitions) {
            Long savedPosition = getSavedPosition(partition);
            if (savedPosition != null) {
                consumer.seek(partition, savedPosition);
            }
        }
    }
});

try {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        try {
            processPayment(record);
            // Сохраняем смещение текущего сообщения
            currentOffset = record.offset();
        } catch (DatabaseException e) {
            // Сохраняем смещение проблемного сообщения и прерываем обработку
            failedOffset = record.offset();
            break;
        }
    }
    
    // Коммитим последнее успешно обработанное сообщение
    if (currentOffset >= 0) {
        consumer.commitSync(Collections.singletonMap(
            new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(currentOffset + 1)
        ));
    }
} catch (Exception e) {
    // Если произошла ошибка, возвращаемся к проблемному смещению
    if (failedOffset >= 0) {
        TopicPartition partition = new TopicPartition("payments", 0);
        consumer.seek(partition, failedOffset);
        // Добавляем экспоненциальную задержку перед повторной попыткой
        Thread.sleep(retryBackoff);
        retryBackoff *= 2; // Увеличиваем задержку для следующей попытки
    }
}

// Создаем отдельных потребителя и производителя
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
 
// Назначаем потребителя на исходный топик
TopicPartition sourcePart = new TopicPartition("source-topic", 0);
consumer.assign(Collections.singletonList(sourcePart));
 
// По умолчанию начинаем с начала топика
consumer.seekToBeginning(Collections.singletonList(sourcePart));
 
// Или с определенной точки
// consumer.seek(sourcePart, startOffset);
 
// Процесс миграции
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    if (records.isEmpty()) {
        // Проверяем, достигли ли мы конечного смещения
        long endOffset = consumer.endOffsets(Collections.singletonList(sourcePart))
                              .get(sourcePart);
        long currentPosition = consumer.position(sourcePart);
        
        if (currentPosition >= endOffset) {
            break; // Миграция завершена
        }
        
        continue;
    }
    
    for (ConsumerRecord<String, String> record : records) {
        // Трансформация данных при необходимости
        String transformedValue = transform(record.value());
        
        // Отправка в целевой топик
        ProducerRecord<String, String> newRecord = 
            new ProducerRecord<>("target-topic", record.key(), transformedValue);
        
        // Добавляем исходные метаданные в заголовки
        newRecord.headers().add("source-topic", record.topic().getBytes());
        newRecord.headers().add("source-partition", 
                               String.valueOf(record.partition()).getBytes());
        newRecord.headers().add("source-offset", 
                               String.valueOf(record.offset()).getBytes());
        
        producer.send(newRecord);
    }
    
    // Периодически сохраняем прогресс миграции
    producer.flush();
    consumer.commitSync();
    
    // Для отслеживания прогресса
    System.out.printf("Мигрировано %d/%d сообщений (%.2f%%)%n", 
                     consumer.position(sourcePart),
                     consumer.endOffsets(Collections.singletonList(sourcePart))
                        .get(sourcePart),
                     (double) consumer.position(sourcePart) * 100 / 
                     consumer.endOffsets(Collections.singletonList(sourcePart))
                        .get(sourcePart));
}

// Класс для хранения информации о запланированной обработке
class ScheduledProcessing {
    private final TopicPartition partition;
    private final long offset;
    private final long scheduledTime;
    
    // Конструктор, геттеры и прочее опущены для краткости
}
 
// Хранилище запланированных обработок
Queue<ScheduledProcessing> scheduledProcessings = new PriorityQueue<>(
    Comparator.comparingLong(ScheduledProcessing::getScheduledTime));
 
// В основном цикле обработки
while (true) {
    // Проверяем, есть ли запланированные операции, время которых настало
    long currentTime = System.currentTimeMillis();
    while (!scheduledProcessings.isEmpty() && 
          scheduledProcessings.peek().getScheduledTime() <= currentTime) {
        ScheduledProcessing sp = scheduledProcessings.poll();
        consumer.seek(sp.getPartition(), sp.getOffset());
        
        // Обрабатываем только одно сообщение из этой позиции
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            if (record.partition() == sp.getPartition().partition() && 
                record.offset() == sp.getOffset()) {
                processRecord(record);
                // Коммитим только это конкретное сообщение
                Map<TopicPartition, OffsetAndMetadata> offsetMap = 
                    Collections.singletonMap(
                        sp.getPartition(),
                        new OffsetAndMetadata(sp.getOffset() + 1)
                    );
                consumer.commitSync(offsetMap);
                break;
            }
        }
    }
    
    // Обычная обработка поступающих сообщений
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        if (shouldProcess(record)) {
            processRecord(record);
        } else {
            // Планируем обработку на будущее
            ScheduledProcessing sp = new ScheduledProcessing(
                new TopicPartition(record.topic(), record.partition()),
                record.offset(),
                calculateProcessingTime(record)
            );
            scheduledProcessings.add(sp);
            
            // Коммитим сообщение, чтобы не получать его снова через poll
            Map<TopicPartition, OffsetAndMetadata> offsetMap = 
                Collections.singletonMap(
                    new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1)
                );
            consumer.commitSync(offsetMap);
        }
    }
}

public class KafkaSeekUtil {
    /**
     * Перемещает консьюмера к заданному смещению во всех назначенных партициях
     */
    public static void seekToOffset(KafkaConsumer<?, ?> consumer, long offset) {
        for (TopicPartition partition : consumer.assignment()) {
            consumer.seek(partition, offset);
        }
    }
    
    /**
     * Перематывает консьюмера на указанное количество сообщений назад
     */
    public static void seekBack(KafkaConsumer<?, ?> consumer, long messagesBack) {
        for (TopicPartition partition : consumer.assignment()) {
            long currentPosition = consumer.position(partition);
            long newPosition = Math.max(0, currentPosition - messagesBack);
            consumer.seek(partition, newPosition);
        }
    }
    
    /**
     * Перемещает к смещению по метке времени
     */
    public static void seekToTime(KafkaConsumer<?, ?> consumer, long timestamp) {
        Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
        for (TopicPartition partition : consumer.assignment()) {
            timestampsToSearch.put(partition, timestamp);
        }
        
        Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes = 
            consumer.offsetsForTimes(timestampsToSearch);
        
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : 
             offsetsForTimes.entrySet()) {
            if (entry.getValue() != null) {
                consumer.seek(entry.getKey(), entry.getValue().offset());
            } else {
                // Если для временной метки нет смещения, 
                // перейти в конец партиции
                consumer.seekToEnd(Collections.singletonList(entry.getKey()));
            }
        }
    }
}

@Test
public void testMessageProcessing() {
    // Подготовка тестовой среды
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
             "org.apache.kafka.common.serialization.StringDeserializer");
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
             "org.apache.kafka.common.serialization.StringDeserializer");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    
    // Создаем тестовый потребитель
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    
    // Заполняем топик тестовыми данными через производителя
    produceTestMessages("test-topic", 10);
    
    // Назначаем потребителя на тестовый топик
    TopicPartition partition = new TopicPartition("test-topic", 0);
    consumer.assign(Collections.singletonList(partition));
    
    // Находим конкретное сообщение для тестирования
    consumer.seekToBeginning(Collections.singletonList(partition));
    ConsumerRecord<String, String> targetRecord = null;
    
    while (targetRecord == null) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            if (record.value().contains("test-condition")) {
                targetRecord = record;
                break;
            }
        }
    }
    
    // Создаем экземпляр тестируемого процессора
    MessageProcessor processor = new MessageProcessor();
    
    // Перематываем к целевому сообщению и проверяем обработку
    consumer.seek(partition, targetRecord.offset());
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        if (record.offset() == targetRecord.offset()) {
            ProcessingResult result = processor.process(record);
            assertEquals(ProcessingResult.SUCCESS, result);
            break;
        }
    }
    
    consumer.close();
}

public class KafkaConsumerMonitor {
    private final KafkaConsumer<String, String> consumer;
    private final Map<TopicPartition, Long> lastProcessedOffsets = new ConcurrentHashMap<>();
    private final Map<TopicPartition, Long> partitionEndOffsets = new ConcurrentHashMap<>();
    
    public KafkaConsumerMonitor(KafkaConsumer<String, String> consumer) {
        this.consumer = consumer;
    }
    
    // Обновить состояние монитора
    public void update() {
        // Получить текущие позиции
        for (TopicPartition partition : consumer.assignment()) {
            lastProcessedOffsets.put(partition, consumer.position(partition));
        }
        
        // Получить конечные смещения
        partitionEndOffsets.putAll(consumer.endOffsets(consumer.assignment()));
    }
    
    // Получить отставание для конкретной партиции
    public long getLag(TopicPartition partition) {
        Long currentEnd = partitionEndOffsets.get(partition);
        Long currentPosition = lastProcessedOffsets.get(partition);
        
        if (currentEnd == null || currentPosition == null) {
            return -1L; // Неизвестно
        }
        
        return currentEnd - currentPosition;
    }
    
    // Получить общее отставание
    public long getTotalLag() {
        return lastProcessedOffsets.keySet().stream()
                .mapToLong(this::getLag)
                .filter(lag -> lag >= 0)
                .sum();
    }
    
    // Перемотать к началу всех партиций
    public void resetToBeginning() {
        consumer.seekToBeginning(consumer.assignment());
        update();
    }
    
    // Перемотать вперед для уменьшения отставания
    public void seekForward(long maxLagThreshold) {
        for (TopicPartition partition : consumer.assignment()) {
            long lag = getLag(partition);
            if (lag > maxLagThreshold) {
                long currentPosition = lastProcessedOffsets.get(partition);
                long newPosition = partitionEndOffsets.get(partition) - maxLagThreshold;
                consumer.seek(partition, Math.max(currentPosition, newPosition));
            }
        }
        update();
    }
}

// Скачок вперед при критическом отставании
if (monitor.getTotalLag() > CRITICAL_LAG_THRESHOLD) {
    // Сохраняем текущие смещения
    Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
    for (TopicPartition partition : consumer.assignment()) {
        currentOffsets.put(partition, 
                          new OffsetAndMetadata(consumer.position(partition)));
    }
    
    // Логируем информацию о пропуске
    log.warn("Критическое отставание: {} сообщений. Пропускаем устаревшие данные", 
            monitor.getTotalLag());
    
    // Переходим к новым смещениям
    Map<TopicPartition, Long> endOffsets = 
        consumer.endOffsets(consumer.assignment());
    for (Map.Entry<TopicPartition, Long> entry : endOffsets.entrySet()) {
        // Оставляем небольшой буфер для обработки
        long newPosition = entry.getValue() - BUFFER_SIZE;
        if (newPosition > consumer.position(entry.getKey())) {
            consumer.seek(entry.getKey(), newPosition);
        }
    }
    
    // Записываем информацию о пропущенном диапазоне
    for (TopicPartition partition : consumer.assignment()) {
        long oldOffset = currentOffsets.get(partition).offset();
        long newOffset = consumer.position(partition);
        long skipped = newOffset - oldOffset;
        
        if (skipped > 0) {
            logSkippedRange(partition, oldOffset, newOffset, skipped);
            sendSkipMetric(partition, skipped);
        }
    }
}

// Класс для хранения информации о повторных попытках
class RetryInfo {
    private final TopicPartition partition;
    private final long offset;
    private int attempts;
    private long nextRetryTime;
    
    // Обновление информации о следующей попытке
    public void incrementAttempt() {
        attempts++;
        // Экспоненциальная задержка: 10ms, 20ms, 40ms, 80ms, ...
        long delay = 10 * (1L << (attempts - 1));
        nextRetryTime = System.currentTimeMillis() + delay;
    }
    
    // Геттеры и прочие методы опущены для краткости
}
 
// Хранилище для сообщений на повторную обработку
Map<String, RetryInfo> messagesToRetry = new HashMap<>();
 
// Основной цикл обработки
while (running) {
    // Обработка запланированных повторных попыток
    long currentTime = System.currentTimeMillis();
    List<String> keysToProcess = messagesToRetry.entrySet().stream()
            .filter(e -> e.getValue().getNextRetryTime() <= currentTime)
            .map(Map.Entry::getKey)
            .collect(Collectors.toList());
    
    for (String key : keysToProcess) {
        RetryInfo info = messagesToRetry.get(key);
        consumer.seek(info.getPartition(), info.getOffset());
        
        // Обработка конкретного сообщения
        ConsumerRecords<String, String> records = 
            consumer.poll(Duration.ofMillis(50));
        for (ConsumerRecord<String, String> record : records) {
            if (record.partition() == info.getPartition().partition() && 
                record.offset() == info.getOffset()) {
                try {
                    processMessage(record);
                    // Успешная обработка — удаляем из списка повторных попыток
                    messagesToRetry.remove(key);
                } catch (ProcessingException e) {
                    // Неудачная попытка — обновляем информацию для следующей попытки
                    info.incrementAttempt();
                    if (info.getAttempts() >= MAX_RETRY_ATTEMPTS) {
                        // Превышено число попыток — отправляем в DLQ
                        sendToDLQ(record, e);
                        messagesToRetry.remove(key);
                    }
                }
                break;
            }
        }
    }
    
    // Обычная обработка новых сообщений
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        try {
            processMessage(record);
        } catch (ProcessingException e) {
            // Добавляем в список повторных попыток
            String key = record.topic() + "-" + record.partition() + "-" + record.offset();
            RetryInfo info = new RetryInfo(
                    new TopicPartition(record.topic(), record.partition()),
                    record.offset());
            info.incrementAttempt();
            messagesToRetry.put(key, info);
        }
    }
}

public static Map<TopicPartition, Long> seekToTimestamp(
        KafkaConsumer<?, ?> consumer, String topic, long timestamp) {
    // Получаем список партиций для топика
    List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
    if (partitionInfos == null || partitionInfos.isEmpty()) {
        throw new RuntimeException("Не найдены партиции для топика: " + topic);
    }
 
    // Создаем список TopicPartition из PartitionInfo
    List<TopicPartition> topicPartitions = partitionInfos.stream()
            .map(info -> new TopicPartition(topic, info.partition()))
            .collect(Collectors.toList());
 
    // Назначаем потребителя на все партиции топика
    consumer.assign(topicPartitions);
 
    // Создаем карту запрашиваемых временных меток
    Map<TopicPartition, Long> timestampsToSearch = topicPartitions.stream()
            .collect(Collectors.toMap(tp -> tp, tp -> timestamp));
 
    // Получаем смещения для указанных временных меток
    Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes = 
            consumer.offsetsForTimes(timestampsToSearch);
 
    // Применяем полученные смещения
    Map<TopicPartition, Long> appliedOffsets = new HashMap<>();
    offsetsForTimes.forEach((tp, offsetAndTimestamp) -> {
        if (offsetAndTimestamp != null) {
            consumer.seek(tp, offsetAndTimestamp.offset());
            appliedOffsets.put(tp, offsetAndTimestamp.offset());
        } else {
            // Если сообщений после указанной временной метки нет,
            // перемещаемся в конец партиции
            consumer.seekToEnd(Collections.singleton(tp));
            long endOffset = consumer.position(tp);
            appliedOffsets.put(tp, endOffset);
        }
    });
 
    return appliedOffsets;
}

public class CheckpointedKafkaConsumer<K, V> implements AutoCloseable {
    private final KafkaConsumer<K, V> consumer;
    private final CheckpointStore checkpointStore;
    private final String consumerId;
    private final String topic;
    
    // Время между сохранениями контрольных точек (мс)
    private final long checkpointIntervalMs;
    private long lastCheckpointTime;
    
    public CheckpointedKafkaConsumer(Properties props, CheckpointStore store, 
                                    String consumerId, String topic) {
        this.consumer = new KafkaConsumer<>(props);
        this.checkpointStore = store;
        this.consumerId = consumerId;
        this.topic = topic;
        this.checkpointIntervalMs = 60000; // 1 минута
        this.lastCheckpointTime = System.currentTimeMillis();
        
        // Инициализация и восстановление из последней контрольной точки
        initialize();
    }
    
    private void initialize() {
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
        List<TopicPartition> topicPartitions = partitionInfos.stream()
                .map(info -> new TopicPartition(topic, info.partition()))
                .collect(Collectors.toList());
        
        consumer.assign(topicPartitions);
        
        // Восстановление из контрольных точек
        Map<TopicPartition, Long> checkpoints = 
                checkpointStore.getCheckpoints(consumerId);
        
        for (TopicPartition partition : topicPartitions) {
            Long checkpoint = checkpoints.get(partition);
            if (checkpoint != null) {
                consumer.seek(partition, checkpoint);
            } else {
                // Для новых партиций начинаем с начала
                consumer.seekToBeginning(Collections.singleton(partition));
            }
        }
    }
    
    public ConsumerRecords<K, V> poll(Duration timeout) {
        ConsumerRecords<K, V> records = consumer.poll(timeout);
        
        // Периодически сохраняем контрольные точки
        long currentTime = System.currentTimeMillis();
        if (currentTime - lastCheckpointTime > checkpointIntervalMs) {
            saveCheckpoints();
            lastCheckpointTime = currentTime;
        }
        
        return records;
    }
    
    public void saveCheckpoints() {
        Map<TopicPartition, Long> positions = new HashMap<>();
        for (TopicPartition partition : consumer.assignment()) {
            positions.put(partition, consumer.position(partition));
        }
        
        checkpointStore.saveCheckpoints(consumerId, positions);
    }
    
    @Override
    public void close() {
        try {
            saveCheckpoints();
        } finally {
            consumer.close();
        }
    }
}

public class MonitoredKafkaConsumer<K, V> implements AutoCloseable {
    private final KafkaConsumer<K, V> consumer;
    private final MetricsRegistry metrics;
    private final String consumerGroup;
    private final ScheduledExecutorService scheduler;
    
    public MonitoredKafkaConsumer(Properties props, MetricsRegistry metrics) {
        this.consumer = new KafkaConsumer<>(props);
        this.metrics = metrics;
        this.consumerGroup = props.getProperty("group.id", "unknown-group");
        
        // Запускаем фоновый мониторинг
        this.scheduler = Executors.newSingleThreadScheduledExecutor();
        this.scheduler.scheduleAtFixedRate(
            this::reportMetrics, 0, 30, TimeUnit.SECONDS);
    }
    
    private void reportMetrics() {
        try {
            // Получаем текущие помещения и лаги партиций
            Map<TopicPartition, Long> currentPositions = new HashMap<>();
            for (TopicPartition partition : consumer.assignment()) {
                currentPositions.put(partition, consumer.position(partition));
            }
            
            // Запрашиваем конечные смешения
            Map<TopicPartition, Long> endOffsets = 
                    consumer.endOffsets(consumer.assignment());
            
            // Рассчитываем и отправляем метрики лага
            for (TopicPartition partition : currentPositions.keySet()) {
                long position = currentPositions.get(partition);
                long endOffset = endOffsets.getOrDefault(partition, position);
                long lag = Math.max(0, endOffset - position);
                
                // Публикуем метрику лага
                metrics.gauge(
                    String.format("kafka.consumer.lag.%s.%s.%d", 
                                 consumerGroup, partition.topic(), 
                                 partition.partition()),
                    lag
                );
                
                // Публикуем метрику текущей позиции
                metrics.gauge(
                    String.format("kafka.consumer.position.%s.%s.%d", 
                                 consumerGroup, partition.topic(), 
                                 partition.partition()),
                    position
                );
            }
        } catch (Exception e) {
            // Логируем ошибку, но не прерываем работу
            log.error("Ошибка при сборе метрик Kafka consumer", e);
        }
    }
    
    // Делегирование основных методов
    public ConsumerRecords<K, V> poll(Duration timeout) {
        return consumer. poll(timeout);
    }
    
    public void seek(TopicPartition partition, long offset) {
        consumer.seek(partition, offset);
        // Обновляем метрики после преднамеренного изменения позиции
        reportMetrics();
    }
    
    @Override
    public void close() {
        try {
            scheduler.shutdown();
            scheduler.awaitTermination(5, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.warn("Ошибка при завершении планировщика", e);
        } finally {
            consumer.close();
        }
    }
}

public class ABTestingConsumer<K, V> {
    private final KafkaConsumer<K, V> consumer;
    private final MessageProcessor processorA;
    private final MessageProcessor processorB;
    private final double samplingRate;
    private final Random random;
    
    public ABTestingConsumer(Properties props, MessageProcessor processorA,
                            MessageProcessor processorB, double samplingRate) {
        this.consumer = new KafkaConsumer<>(props);
        this.processorA = processorA;
        this.processorB = processorB;
        this.samplingRate = samplingRate;
        this.random = new Random();
    }
    
    public void processMessages() {
        while (true) {
            ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));
            if (records.isEmpty()) continue;
            
            // Сохраняем текущее положение для каждой партиции
            Map<TopicPartition, Long> savedPositions = new HashMap<>();
            for (TopicPartition partition : consumer.assignment()) {
                savedPositions.put(partition, consumer.position(partition));
            }
            
            // Обрабатываем записи с процессором A
            for (ConsumerRecord<K, V> record : records) {
                processorA.process(record);
            }
            
            // Решаем, нужно ли запускать тестовый вариант B
            if (random.nextDouble() < samplingRate) {
                // Возвращаемся к исходной позиции для повторной обработки
                savedPositions.forEach(consumer::seek);
                
                // Обрабатываем те же записи с процессором B и сравниваем результаты
                for (ConsumerRecord<K, V> record : records) {
                    ComparisonResult result = new ComparisonResult(
                        processorA.getLastResult(record),
                        processorB.process(record)
                    );
                    
                    // Анализируем различия
                    analyzeAndReportComparison(record, result);
                }
            }
            
            // Коммитим обработанные сообщения
            consumer.commitSync();
        }
    }
}

public class PartialCommitConsumer<K, V> {
    private final KafkaConsumer<K, V> consumer;
    private final Map<String, Set<TopicPartition>> completedBatches = new HashMap<>();
    
    public void processBatchWithPartialCommits(String batchId, int totalExpectedRecords) {
        // Схема обработки сообщений с частичными фиксациями
        int processedRecords = 0;
        boolean batchComplete = false;
        
        // Партиции, для которых достигнуты условия коммита
        Set<TopicPartition> completedPartitions = 
                completedBatches.computeIfAbsent(batchId, k -> new HashSet<>());
        
        while (!batchComplete) {
            ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));
            Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
            
            for (ConsumerRecord<K, V> record : records) {
                TopicPartition partition = 
                        new TopicPartition(record.topic(), record.partition());
                
                // Пропускаем уже обработанные партиции
                if (completedPartitions.contains(partition)) continue;
                
                // Обработка сообщения
                processRecord(record);
                processedRecords++;
                
                // Обновляем смещения для коммита
                offsetsToCommit.put(
                    partition,
                    new OffsetAndMetadata(record.offset() + 1, batchId)
                );
                
                // Проверяем критерии завершения пакета
                if (isPartitionComplete(partition, batchId)) {
                    completedPartitions.add(partition);
                    
                    // Сохраняем прогресс обработки для конкретной партиции
                    consumer.commitSync(
                        Collections.singletonMap(
                            partition,
                            new OffsetAndMetadata(record.offset() + 1, 
                                                 batchId + ":completed")
                        )
                    );
                }
            }
            
            // Коммитим промежуточный прогресс для активных партиций
            if (!offsetsToCommit.isEmpty()) {
                consumer.commitSync(offsetsToCommit);
            }
            
            // Проверяем общее завершение пакета
            batchComplete = processedRecords >= totalExpectedRecords || 
                            completedPartitions.containsAll(consumer.assignment());
        }
        
        // Очистка
        completedBatches.remove(batchId);
    }
    
    // Перезапуск обработки с учётом данных частичного коммита
    public void resumeBatch(String batchId) {
        Set<TopicPartition> partitionsToProcess = new HashSet<>(consumer.assignment());
        
        // Проверяем, какие партиции уже полностью обработаны
        for (TopicPartition partition : consumer.assignment()) {
            OffsetAndMetadata metadata = consumer.committed(partition);
            if (metadata != null && metadata.metadata().contains(batchId + ":completed")) {
                partitionsToProcess.remove(partition);
                completedBatches.computeIfAbsent(batchId, k -> new HashSet<>())
                               .add(partition);
            }
        }
        
        // Продолжаем обработку только для незавершённых партиций
        // Для них позиции уже установлены на последние коммитированные
    }
}

// Группировка операций seek для снижения накладных расходов
public void optimizedSeeks(KafkaConsumer<String, String> consumer,
                         Map<TopicPartition, Long> targetOffsets) {
    // Сгруппируем операции по топику и партиции
    Map<String, List<TopicPartition>> topicPartitions = new HashMap<>();
    
    for (TopicPartition tp : targetOffsets.keySet()) {
        topicPartitions.computeIfAbsent(tp.topic(), k -> new ArrayList<>())
                      .add(tp);
    }
    
    // Выполняем операции seek группами для каждого топика
    for (Map.Entry<String, List<TopicPartition>> entry : topicPartitions.entrySet()) {
        for (TopicPartition tp : entry.getValue()) {
            consumer.seek(tp, targetOffsets.get(tp));
        }
        
        // Одна операция poll после группы seek для одного топика
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(0));
        // Игнорируем результат - это просто для применения операций seek
    }
}

// Проблемный код
consumer.seekToBeginning(consumer.assignment());
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
// Предполагается, что здесь будут записи с начала партиции, но...
// Если в буфере потребителя уже были записи из предыдущего poll(),
// они будут возвращены ПЕРЕД тем, как будет выполнен seek!
 
// Исправленный вариант
consumer.seekToBeginning(consumer.assignment());
// Сначала выполняем poll с нулевым таймаутом для применения seek
consumer.poll(Duration.ofMillis(0));
// Теперь получаем записи с новой позиции
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

// Проблемный код с автоматическим коммитом
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
// ...
consumer.seek(partition, specificOffset);
// Обработка...
// Проблема: автоматический коммит может произойти в любой момент,
// перезаписав смещение, которое мы установили через seek
 
// Решение: отключить автоматический коммит и управлять вручную
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
// ...
consumer.seek(partition, specificOffset);
// Обработка...
// Явный коммит только когда это нужно
consumer.commitSync();

public class SeekPerformanceAnalyzer {
    public static void analyzePartitioningImpact(String bootstrapServers, String topic,
                                              int[] partitionCounts) throws Exception {
        for (int partitionCount : partitionCounts) {
            // Создаем топик с заданным количеством партиций
            AdminClient admin = AdminClient.create(
                Collections.singletonMap(
                    AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers
                )
            );
            
            NewTopic newTopic = new NewTopic(
                topic + "-" + partitionCount,
                partitionCount,
                (short) 1
            );
            
            admin.createTopics(Collections.singleton(newTopic)).all().get();
            admin.close();
            
            // Заполняем данными
            populateTestData(bootstrapServers, topic + "-" + partitionCount, 100000);
            
            // Тестируем seek-операции
            long startTime = System.currentTimeMillis();
            testSeekOperations(bootstrapServers, topic + "-" + partitionCount, 100);
            long endTime = System.currentTimeMillis();
            
            System.out.printf(
                "Топик с %d партициями: время выполнения seek-операций: %d ms%n",
                partitionCount, (endTime - startTime)
            );
        }
    }
    
    private static void testSeekOperations(String bootstrapServers, 
                                        String topic, int seekOperations) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "seek-performance-test");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
                 StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
                 StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            List<PartitionInfo> partitions = consumer.partitionsFor(topic);
            List<TopicPartition> topicPartitions = partitions.stream()
                    .map(p -> new TopicPartition(topic, p.partition()))
                    .collect(Collectors.toList());
            
            consumer.assign(topicPartitions);
            
            Random random = new Random();
            for (int i = 0; i < seekOperations; i++) {
                // Выбираем случайную партицию
                TopicPartition tp = topicPartitions.get(
                    random.nextInt(topicPartitions.size())
                );
                
                // Получаем границы смещений
                Map<TopicPartition, Long> beginningOffsets = 
                    consumer.beginningOffsets(Collections.singleton(tp));
                Map<TopicPartition, Long> endOffsets = 
                    consumer.endOffsets(Collections.singleton(tp));
                
                long start = beginningOffsets.get(tp);
                long end = endOffsets.get(tp);
                
                // Выполняем seek к случайной позиции
                if (end > start) {
                    long randomOffset = start + random.nextInt((int)(end - start));
                    consumer.seek(tp, randomOffset);
                    consumer.poll(Duration.ofMillis(0));
                }
            }
        }
    }
}

public class CoordinatedSeekManager {
    private final StateStore stateStore;
    private final String applicationId;
    
    public CoordinatedSeekManager(StateStore stateStore, String applicationId) {
        this.stateStore = stateStore;
        this.applicationId = applicationId;
    }
    
    // Сохранение позиции с возможностью атомарного обновления
    public void savePosition(String topic, int partition, long offset, 
                          String metadata) throws Exception {
        String key = formatKey(topic, partition);
        PositionInfo current = stateStore.getPosition(applicationId, key);
        
        // Обновляем только если позиция новее
        if (current == null || offset > current.offset) {
            PositionInfo newPosition = new PositionInfo(offset, metadata, 
                                                      System.currentTimeMillis());
            stateStore.updatePosition(applicationId, key, newPosition);
        }
    }
    
    // Загрузка координированной позиции
    public PositionInfo loadPosition(String topic, int partition) throws Exception {
        String key = formatKey(topic, partition);
        return stateStore.getPosition(applicationId, key);
    }
    
    // Координированный seek для группы потребителей
    public void coordinatedSeek(KafkaConsumer<?, ?> consumer, 
                               Collection<TopicPartition> partitions) throws Exception {
        for (TopicPartition tp : partitions) {
            PositionInfo position = loadPosition(tp.topic(), tp.partition());
            if (position != null) {
                consumer.seek(tp, position.offset);
            }
        }
    }
    
    private String formatKey(String topic, int partition) {
        return topic + "-" + partition;
    }
    
    // Класс для хранения информации о позиции
    public static class PositionInfo {
        public final long offset;
        public final String metadata;
        public final long timestamp;
        
        public PositionInfo(long offset, String metadata, long timestamp) {
            this.offset = offset;
            this.metadata = metadata;
            this.timestamp = timestamp;
        }
    }
    
    // Интерфейс для хранилища состояния
    public interface StateStore {
        PositionInfo getPosition(String appId, String key) throws Exception;
        void updatePosition(String appId, String key, PositionInfo position) 
                          throws Exception;
    }
}

public class KafkaBasedCoordinationStore implements CoordinatedSeekManager.StateStore {
    private final KafkaProducer<String, String> producer;
    private final KafkaConsumer<String, String> consumer;
    private final String coordinationTopic;
    private final ObjectMapper objectMapper;
    
    public KafkaBasedCoordinationStore(Properties producerProps, 
                                     Properties consumerProps,
                                     String coordinationTopic) {
        this.producer = new KafkaProducer<>(producerProps);
        this.consumer = new KafkaConsumer<>(consumerProps);
        this.coordinationTopic = coordinationTopic;
        this.objectMapper = new ObjectMapper();
        
        // Подписываемся на топик координации
        consumer.subscribe(Collections.singleton(coordinationTopic));
    }
    
    @Override
    public CoordinatedSeekManager.PositionInfo getPosition(String appId, String key) 
                                                        throws Exception {
        // Формируем полный ключ для поиска
        String fullKey = appId + ":" + key;
        
        // Переходим к последнему смещению
        consumer.seekToEnd(consumer.assignment());
        
        // Читаем последние N сообщений (можно настроить)
        long startTime = System.currentTimeMillis();
        Map<String, CoordinatedSeekManager.PositionInfo> latestPositions = 
                new HashMap<>();
        
        while (System.currentTimeMillis() - startTime < 5000) { // Таймаут 5 секунд
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            
            for (ConsumerRecord<String, String> record : records) {
                // Ищем записи с нашим ключом
                if (record.key().equals(fullKey)) {
                    CoordinatedSeekManager.PositionInfo position = 
                            objectMapper.readValue(record.value(), 
                                                 CoordinatedSeekManager.PositionInfo.class);
                    
                    // Сохраняем только самую свежую версию
                    if (!latestPositions.containsKey(fullKey) || 
                        position.timestamp > latestPositions.get(fullKey).timestamp) {
                        latestPositions.put(fullKey, position);
                    }
                }
            }
            
            // Если нашли позицию, можно прервать поиск
            if (latestPositions.containsKey(fullKey)) {
                break;
            }
        }
        
        return latestPositions.get(fullKey);
    }
    
    @Override
    public void updatePosition(String appId, String key, 
                            CoordinatedSeekManager.PositionInfo position) 
                            throws Exception {
        // Формируем полный ключ
        String fullKey = appId + ":" + key;
        
        // Сериализуем позицию в JSON
        String positionJson = objectMapper.writeValueAsString(position);
        
        // Отправляем в топик координации
        ProducerRecord<String, String> record = 
                new ProducerRecord<>(coordinationTopic, fullKey, positionJson);
        
        // Ждем подтверждения записи для гарантии доставки
        producer.send(record).get();
    }
    
    public void close() {
        producer.close();
        consumer.close();
    }
}

public class LoadBalancedConsumerGroup implements Runnable {
    private final List<KafkaConsumer<String, String>> consumers;
    private final int totalPartitions;
    private final AtomicLong[] partitionLags;
    private final ReentrantLock rebalanceLock = new ReentrantLock();
    private final ScheduledExecutorService scheduler = 
        Executors.newSingleThreadScheduledExecutor();
    
    public LoadBalancedConsumerGroup(List<Properties> consumerConfigs, 
                                  String topic, int partitionCount) {
        this.consumers = consumerConfigs.stream()
                .map(props -> new KafkaConsumer<String, String>(props))
                .collect(Collectors.toList());
        
        this.totalPartitions = partitionCount;
        this.partitionLags = new AtomicLong[partitionCount];
        
        for (int i = 0; i < partitionCount; i++) {
            partitionLags[i] = new AtomicLong(0);
        }
        
        // Запускаем периодическую ребалансировку
        scheduler.scheduleAtFixedRate(
            this::rebalancePartitions, 30, 30, TimeUnit.SECONDS);
    }
    
    @Override
    public void run() {
        // Запускаем потребителей в отдельных потоках
        List<Thread> consumerThreads = new ArrayList<>();
        
        for (int i = 0; i < consumers.size(); i++) {
            final int consumerId = i;
            Thread thread = new Thread(() -> consumeMessages(consumerId));
            consumerThreads.add(thread);
            thread.start();
        }
        
        // Ожидаем завершения всех потоков
        for (Thread thread : consumerThreads) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    
    private void consumeMessages(int consumerId) {
        KafkaConsumer<String, String> consumer = consumers.get(consumerId);
        
        // Начальное назначение партиций (равномерно)
        assignInitialPartitions(consumer, consumerId);
        
        while (true) {
            try {
                ConsumerRecords<String, String> records = 
                    consumer.poll(Duration.ofMillis(100));
                
                // Обработка записей
                for (ConsumerRecord<String, String> record : records) {
                    processRecord(record);
                    
                    // Обновляем информацию об отставании
                    int partition = record.partition();
                    partitionLags[partition].decrementAndGet();
                }
                
                // Обновляем информацию о лаге для назначенных партиций
                updateLagInfo(consumer);
                
                // Проверка на блокировку ребалансировки
                if (rebalanceLock.isLocked() && 
                    rebalanceLock.isHeldByCurrentThread()) {
                    // Выполняем специфичные для ребалансировки операции
                    handleRebalance(consumer, consumerId);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    
    private void assignInitialPartitions(KafkaConsumer<String, String> consumer, 
                                      int consumerId) {
        int consumerCount = consumers.size();
        List<TopicPartition> partitionsToAssign = new ArrayList<>();
        
        // Распределяем партиции равномерно
        for (int i = consumerId; i < totalPartitions; i += consumerCount) {
            partitionsToAssign.add(new TopicPartition("test-topic", i));
        }
        
        consumer.assign(partitionsToAssign);
        consumer.seekToEnd(partitionsToAssign);
    }
    
    private void updateLagInfo(KafkaConsumer<String, String> consumer) {
        Set<TopicPartition> assignment = consumer.assignment();
        
        // Получаем конечные смещения
        Map<TopicPartition, Long> endOffsets = consumer.endOffsets(assignment);
        
        // Обновляем информацию о лаге
        for (TopicPartition tp : assignment) {
            long currentPosition = consumer.position(tp);
            long endOffset = endOffsets.get(tp);
            long lag = Math.max(0, endOffset - currentPosition);
            
            partitionLags[tp.partition()].set(lag);
        }
    }
    
    private void rebalancePartitions() {
        try {
            // Блокируем доступ к ребалансировке
            rebalanceLock.lock();
            
            // Анализируем текущее распределение нагрузки
            long[] consumerLoads = new long[consumers.size()];
            Map<Integer, Set<Integer>> consumerToPartitions = new HashMap<>();
            
            for (int i = 0; i < consumers.size(); i++) {
                consumerToPartitions.put(i, new HashSet<>());
            }
            
            KafkaConsumer<String, String> firstConsumer = consumers.get(0);
            for (int i = 0; i < totalPartitions; i++) {
                TopicPartition tp = new TopicPartition("test-topic", i);
                
                // Определяем, какому потребителю назначена эта партиция
                for (int j = 0; j < consumers.size(); j++) {
                    KafkaConsumer<String, String> consumer = consumers.get(j);
                    if (consumer.assignment().contains(tp)) {
                        consumerToPartitions.get(j).add(i);
                        consumerLoads[j] += partitionLags[i].get();
                        break;
                    }
                }
            }
            
            // Находим наиболее и наименее загруженных потребителей
            int maxLoadConsumer = 0;
            int minLoadConsumer = 0;
            
            for (int i = 1; i < consumers.size(); i++) {
                if (consumerLoads[i] > consumerLoads[maxLoadConsumer]) {
                    maxLoadConsumer = i;
                }
                if (consumerLoads[i] < consumerLoads[minLoadConsumer]) {
                    minLoadConsumer = i;
                }
            }
            
            // Если дисбаланс достаточно большой, перераспределяем партиции
            if (consumerLoads[maxLoadConsumer] > consumerLoads[minLoadConsumer] * 1.5) {
                // Находим партицию с наибольшим лагом у самого загруженного потребителя
                int partitionToMove = -1;
                long maxLag = -1;
                
                for (int partitionId : consumerToPartitions.get(maxLoadConsumer)) {
                    if (partitionLags[partitionId].get() > maxLag) {
                        maxLag = partitionLags[partitionId].get();
                        partitionToMove = partitionId;
                    }
                }
                
                if (partitionToMove >= 0) {
                    // Переносим партицию от максимально к минимально загруженному
                    TopicPartition tp = new TopicPartition("test-topic", partitionToMove);
                    
                    // Сохраняем текущую позицию
                    KafkaConsumer<String, String> sourceConsumer = 
                        consumers.get(maxLoadConsumer);
                    long currentPosition = sourceConsumer.position(tp);
                    
                    // Удаляем из назначения источника
                    Set<TopicPartition> newSourceAssignment = new HashSet<>(
                        sourceConsumer.assignment());
                    newSourceAssignment.remove(tp);
                    sourceConsumer.assign(newSourceAssignment);
                    
                    // Добавляем в назначение приемника
                    KafkaConsumer<String, String> targetConsumer = 
                        consumers.get(minLoadConsumer);
                    Set<TopicPartition> newTargetAssignment = new HashSet<>(
                        targetConsumer.assignment());
                    newTargetAssignment.add(tp);
                    targetConsumer.assign(newTargetAssignment);
                    
                    // Позиционируем на той же позиции
                    targetConsumer.seek(tp, currentPosition);
                    
                    System.out.printf(
                        "Перенесена партиция %d от потребителя %d к потребителю %d%n",
                        partitionToMove, maxLoadConsumer, minLoadConsumer
                    );
                }
            }
        } finally {
            rebalanceLock.unlock();
        }
    }
    
    private void handleRebalance(KafkaConsumer<String, String> consumer, 
                               int consumerId) {
        // Здесь может быть реализована дополнительная логика 
        // при ребалансировке
    }
    
    private void processRecord(ConsumerRecord<String, String> record) {
        // Логика обработки сообщения
    }
}

public class ThrottledConsumer<K, V> {
    private final KafkaConsumer<K, V> consumer;
    private final long maxMessagesPerSecond;
    private long messagesProcessedInCurrentSecond = 0;
    private long currentSecondStartTime = System.currentTimeMillis();
    
    public ThrottledConsumer(KafkaConsumer<K, V> consumer, long maxMessagesPerSecond) {
        this.consumer = consumer;
        this.maxMessagesPerSecond = maxMessagesPerSecond;
    }
    
    public ConsumerRecords<K, V> throttledPoll(Duration timeout) {
        updateThrottlingState();
        
        if (messagesProcessedInCurrentSecond >= maxMessagesPerSecond) {
            // Достигли лимита - возвращаем пустой результат
            return ConsumerRecords.empty();
        }
        
        // Оставшееся количество сообщений, которое можно обработать в этой секунде
        long remainingQuota = maxMessagesPerSecond - messagesProcessedInCurrentSecond;
        
        ConsumerRecords<K, V> records = consumer.poll(timeout);
        int recordCount = countRecords(records);
        
        if (recordCount <= remainingQuota) {
            // Можем обработать все записи
            messagesProcessedInCurrentSecond += recordCount;
            return records;
        } else {
            // Нужно ограничить количество записей
            messagesProcessedInCurrentSecond += remainingQuota;
            
            // Обрабатываем только часть записей, и перемещаемся к позициям
            // для последующего потребления
            Map<TopicPartition, List<ConsumerRecord<K, V>>> filteredRecords = 
                new HashMap<>();
            Map<TopicPartition, Long> newPositions = new HashMap<>();
            
            long processed = 0;
            for (TopicPartition partition : records.partitions()) {
                List<ConsumerRecord<K, V>> partitionRecords = 
                    records.records(partition);
                
                if (processed + partitionRecords.size() <= remainingQuota) {
                    // Можем взять все записи из этой партиции
                    filteredRecords.put(partition, new ArrayList<>(partitionRecords));
                    processed += partitionRecords.size();
                } else {
                    // Берем только часть записей
                    long canTake = remainingQuota - processed;
                    List<ConsumerRecord<K, V>> limitedRecords = 
                        partitionRecords.subList(
                            0, (int) canTake);
                    
                    filteredRecords.put(partition, 
                                      new ArrayList<>(limitedRecords));
                    
                    // Запоминаем позицию для последующего чтения
                    newPositions.put(partition, 
                                   limitedRecords.get(
                                       limitedRecords.size() - 1).offset() + 1);
                    
                    processed += canTake;
                    
                    // Прерываем цикл, если квота исчерпана
                    if (processed >= remainingQuota) {
                        break;
                    }
                }
            }
            
            // Устанавливаем позиции для следующего чтения
            for (Map.Entry<TopicPartition, Long> entry : newPositions.entrySet()) {
                consumer.seek(entry.getKey(), entry.getValue());
            }
            
            return new ConsumerRecords<>(filteredRecords);
        }
    }
    
    private void updateThrottlingState() {
        long currentTime = System.currentTimeMillis();
        long elapsedMs = currentTime - currentSecondStartTime;
        
        if (elapsedMs >= 1000) {
            // Прошла секунда - сбрасываем счетчик
            currentSecondStartTime = currentTime;
            messagesProcessedInCurrentSecond = 0;
        }
    }
    
    private int countRecords(ConsumerRecords<K, V> records) {
        int count = 0;
        for (TopicPartition partition : records.partitions()) {
            count += records.records(partition).size();
        }
        return count;
    }
}