Кафка не потребляет сообщения

Question

Я обновил свой kafka с версии 0.10.2.0 до версии 2.1.0, и теперь kafka не может принимать сообщения. Я использую весеннюю загрузку, и вот моя конфигурация:

configProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServers);
configProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
configProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
configProperties.put(ConsumerConfig.GROUP_ID_CONFIG, topic + "kafka2");
configProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
configProperties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 3000000);
configProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
configProperties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 2000000000);
configProperties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "900000");
configProperties.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "905000");
configProperties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "35000");

Я изменил идентификатор группы, чтобы избежать проблем со старым идентификатором группы. Моя текущая версия весны - 2.1.2. И в моем приложении я вижу, как мой клиент постоянно переподключается

Есть идеи о проблеме?

Answer 1

Вы должны быть более конкретными, когда говорите о "весенних версиях".Существует множество весенних проектов с разными версиями.

Полагаю, вы имеете в виду Spring Boot 2.1.2 (кстати, текущая версия 2.1.x - 2.1.6).

Spring Boot2.1.6 использует Spring для Apache Kafka 2.2.7, который, в свою очередь, использует kafka-clients 2.0.1.

. Я не ожидал бы никаких проблем при использовании клиентов kafka 2.0.x с брокером 2.1.0., но вы можете попробовать переопределить версии клиентов kafka до 2.1.0, как описано в справочном руководстве .

. Снова, я предлагаю включить ведение журнала DEBUG для apache.kafka, чтобы проверить, обеспечивает ли оноВы с дополнительной информацией.