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 в качестве ключа для каждого сообщения.

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими пользовательским соглашением, политикой конфиденциальности и политикой о куки

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