Автоматическое удаление группы потребителей кафки после отключения

Question

В моем приложении Java Раз в пару секунд я назначаю потребителю определенный c TopicPartition и пытаюсь прочитать указанное c сообщение из указанного c Топи c + раздел. После прочтения сообщения (используя poll () ), я немедленно отключаю потребителя.

Поскольку приведенный выше сценарий может работать в многопоточной среде, имя группы потребителей имеет префикс + случайный ха sh Например, my_consumer_group_EWQSV (поскольку kafka не назначит один и тот же указанный c раздел двум пользователям в одной группе).

Проблема в том, что я не смог Скажите kafka, чтобы удалить этих потребителей после их отключения (так как эти потребители просто временные), есть ли способ сделать это? (не вручную, я имею в виду, используя конфигурацию или что-то в этом роде, я не смог найти никакой конф, такой как «автоматическое удаление-после-отключения-отсоединения»)

Спасибо:)

Answer 1

В java, для того чтобы вручную потреблять заданное c сообщение от kafka без создания группы потребителей:

Достаточно следующее:

kafkaProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        kafkaProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        TopicPartition tp = new TopicPartition(topic, partition);
        try (Consumer<String, String> consumer = new KafkaConsumer<>(kafkaProps)) {
            consumer.assign(Collections.singletonList(tp));
            consumer.seek(tp, offset);
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));
        } 

Важные части:

• установить для свойства enable.auto.commit значение false (как показано выше).
• do NOT установить group.id ( ConsumerConfig.GROUP_ID_CONFIG ) свойство, поскольку идентификатор не нужен.
• do NOT использовать подписку метод, который автоматически назначает разделы, вместо этого используйте методы assign и seek для ручного чтения сообщения как показано выше.