При работе с Apache Kafka метод onPartitionsAssigned играет решающую роль в процессе координации группы потребителей. В этой статье блога мы рассмотрим обратный вызов onPartitionsAssigned и его значение для эффективной обработки назначения разделов. Мы углубимся в примеры кода и будем использовать разговорный язык, чтобы сделать эту тему доступной для всех читателей. Итак, начнем!

Метод 1: базовая реализация

Самый простой способ использовать обратный вызов onPartitionsAssigned — реализовать интерфейс ConsumerRebalanceListener. Вот пример того, как этого можно добиться на Java:

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.common.TopicPartition;
import java.util.Collection;
public class MyConsumerRebalanceListener implements ConsumerRebalanceListener {
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Your logic to handle partition assignments goes here
        System.out.println("Partitions assigned: " + partitions);
        // Additional code to handle the assigned partitions
    }
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Your logic to handle partition revocations goes here
        System.out.println("Partitions revoked: " + partitions);
        // Additional code to handle the revoked partitions
    }
}
// Create a Kafka consumer and attach the listener
consumer.subscribe(Collections.singletonList("my_topic"), new MyConsumerRebalanceListener());

Метод 2: Стратегия индивидуального назначения разделов

В некоторых случаях вам может потребоваться больше контроля над назначением разделов. Kafka позволяет реализовать собственную стратегию назначения разделов путем расширения класса AbstractPartitionAssignor. Вот пример на Java:

import org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor;
import org.apache.kafka.clients.consumer.internals.PartitionAssignor;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.protocol.types.Schema;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.Set;
public class CustomPartitionAssignor extends AbstractPartitionAssignor {
    @Override
    public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
                                                    Map<String, Subscription> subscriptions) {
        // Your custom logic to assign partitions goes here
        // Return a map of assigned partitions per consumer
    }
    @Override
    public ByteBuffer subscriptionUserData(Set<String> topics) {
        // Your custom logic to encode subscription data goes here
        // Return a byte buffer representing the encoded data
    }
    @Override
    public void onAssignment(Assignment assignment, ConsumerGroupMetadata metadata) {
        // Your logic to handle the assignment result goes here
        // Additional code to handle the assigned partitions
    }
}
// Create a Kafka consumer and assign the custom partition assignor
Properties props = new Properties();
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, CustomPartitionAssignor.class.getName());
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my_topic"));

Заключение

Обратный вызов onPartitionsAssigned — это мощный инструмент в Apache Kafka для обработки назначений разделов в группах потребителей. Реализуя интерфейс ConsumerRebalanceListener или расширяя класс AbstractPartitionAssignor, вы можете настроить поведение своего потребительского приложения для обработки назначений разделов в соответствии с вашими требованиями.

В этой статье мы рассмотрели два метода использования onPartitionsAssigned на примерах кода. Независимо от того, выберете ли вы базовую реализацию или стратегию назначения настраиваемых разделов, понимание и использование этого обратного вызова значительно повысят производительность и гибкость вашего потребителя Kafka.