Кафка Потоки RoundRobinPartitioner - PullRequest
Новая
       4

Кафка Потоки RoundRobinPartitioner

1 голос
/ 08 января 2020

Я написал код потоков kafka, который использует версию клиента kafka 2.4 kafka и версию сервера kafka 2.2. У меня 50 разделов на моем topi c и внутреннем topi c.

В моем коде потока kafka есть операция selectKey () DSL, и у меня 2 миллиона записей с использованием того же KEY. В конфигурации потока я сделал

props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, RoundRobinPartitioner.class);

, чтобы иметь возможность использовать разные разделы с одинаковым ключом. Если я не использую Round Robin, как ожидалось, все мои сообщения go в один и тот же раздел.

Пока все в порядке, но я это понял; когда я использую класс RoundRobinPartitioner, мои сообщения go напоминают ~ 40 разделов. 10 раздел находится в нерабочем состоянии. Интересно, что мне не хватает? Он должен использовать 50 из них около 2 миллионов записей, верно? 

      final KStream<String, IdListExportMessage> exportedDeviceIdsStream =
            builder.stream("deviceIds");

        // k: appId::deviceId, v: device
        final KTable<String, Device> deviceTable = builder.table(
            "device",
            Consumed.with(Serdes.String(), deviceSerde)
        );
            // Some DSL operations
            .join(
                deviceTable,
                (exportedDevice, device) -> {
                    exportedDevice.setDevice(device);

                    return exportedDevice;
                },
                Joined.with(Serdes.String(), exportedDeviceSerde, deviceSerde)
            )
            .selectKey((deviceId, exportedDevice) -> exportedDevice.getDevice().getId())
            .to("bulk_consumer");

И 

   props.put(StreamsConfig.STATE_DIR_CONFIG, /tmp/kafka-streams);
   props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
   props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 2);
   props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);
   props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, EXACTLY_ONCE);
   props.put("num.stream.threads", 10);
   props.put("application.id", applicationId);

RoundRobinPartitioner. java

public class RoundRobinPartitioner implements Partitioner {
    private final ConcurrentMap<String, AtomicInteger> topicCounterMap = new ConcurrentHashMap();

    public RoundRobinPartitioner() {
    }

    public void configure(Map<String, ?> configs) {
    }

    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        int nextValue = this.nextValue(topic);
        List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
        if (!availablePartitions.isEmpty()) {
            int part = Utils.toPositive(nextValue) % availablePartitions.size();
            return ((PartitionInfo)availablePartitions.get(part)).partition();
        } else {
            return Utils.toPositive(nextValue) % numPartitions;
        }
    }

    private int nextValue(String topic) {
        AtomicInteger counter = (AtomicInteger)this.topicCounterMap.computeIfAbsent(topic, (k) -> {
            return new AtomicInteger(0);
        });
        return counter.getAndIncrement();
    }

    public void close() {
    }
}

1 Ответ

1 голос
/ 09 января 2020

Вы не можете изменить разбиение с помощью конфигурации ProducerConfig.PARTITIONER_CLASS_CONFIG - это работает только для простого производителя.

В Kafka Streams вам необходимо реализовать интерфейс StreamsPartitioner и передать реализацию в соответствующие операторы, например, to("topic", Produced.streamPartitioner(new MyPartitioner()).

...