0

У меня есть топик first_topic с 3 partitions.

enter image description here

Вот мой код, я отправляю 55 раз сообщение косьюмеру ( который у меня открыт в cmd ), код ниже прописывает в консоли partition, куда было отправлено сообщение.

Но вот проблема: каждый раз, когда я запускаю этот код, - все 55 сообщений уходят в случайный 1 партишен. Я думал, что если key не указан, то должно быть round-robin ( каждый партишен должен получить понемногу ). Почему в данном примере это не работает?

   Logger logger = LoggerFactory.getLogger(Producer.class);
    Properties properties = new Properties();
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());


    KafkaProducer<String, String> producer = new KafkaProducer(properties);


    for (int i = 0; i < 55; i++) {


    producer.send(new ProducerRecord<>("first_topic", "Hello world partitionCheck " + i), (recordMetadata, e) -> {
        // executes every time a record is sent or Exception is thrown
        if (e == null) {
            // record was successfully sent
            logger.info("metaData: " +
                    "topic " + recordMetadata.topic() +
                    " Offset " + recordMetadata.offset() +
                    " TimeStamp " + recordMetadata.timestamp() +
                    " Partition " + recordMetadata.partition());
        } else {
            logger.error(e.toString());
        }
    });
    }

    producer.flush();
    producer.close();

1 ответ 1

0

Документация у них слегка отстаёт.

Когда нет ключа и нет custom partitioner, то управление передаётся в BuiltInPartitioner.

Этот класс переключается между партициями только когда в партицию отправлено определённое количество байт (задается параметром batch.size).

https://github.com/a0x8o/kafka/blob/master/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java#L264

https://github.com/a0x8o/kafka/blob/master/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java#L138

https://github.com/a0x8o/kafka/blob/master/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java#L285

https://github.com/a0x8o/kafka/blob/master/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java#L172


Если хочешь, чтобы переключалось на каждом сообщении, то просто генерируй UUID в качестве ключа для каждого сообщения.

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.