Если вы хотите контролировать скорость, с которой потребитель Kafka использует Spring @KafkaListener, пожалуйста, автоматически подключите использование компонента KafkaListenerEndpointRegistry следующим образом и получите доступ к необходимому MessageListenerContainer. после этого вы можете использовать функции pause () и resume () для управления требуемым поведением.

тели использовать записи через определенный промежуток времени (например, каждые 5 минут). Потребительские свойства являются стандартными: @Bean public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> ...

Я думаю, что вы можете использовать что-то вроде этого:

ли способ сделать операцию, подобную ветви, но поместить запись в каждый выходной поток, предикат которого оценивается как true? Brach помещает запись в первое совпадение (документация: запись помещается в один и только один выходной поток при ...

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

 будет использовать нового потребителя в этом случае (но вы не можете видеть заголовки с консольным потребителем, в любом случае). Но я согласен, что странно, что потребитель может повлиять на производителя. Возможно, есть какая-то логика, которая гласит: «У нас есть старый потребитель, присоединенный к этой теме, поэтому вы не можете отправлять заголовки»

ользую эту настройку docker-compose для локальной настройки Kafka: https://github.com/wurstmeister/kafka-docker/ [https://github.com/wurstmeister/kafka-docker/] docker-compose up работает отлично, создание тем через оболочку ...

Чтение того же сообщения несколько раз от Кафки

Спасибо, Гэри. Ты спас мой день :)

работал@KafkaListener это также отмеченоConsumerAwareRebalanceListener интерфейс, используя Spring Boot 2.0.6. Я реализовалonPartitionsAssigned метод, в котором я перематываю смещение на фиксированное количество времени, скажем, 60 секунд. Все ...

github.com/Yolean/kubernetes-kafka

ольно новичок в Kubernetes и хотел настроить Kafka и zookeeper с ним. Я смог настроить Apache Kafka и Zookeeper в Kubernetes, используя StatefulSets. Я последовал заэто [https://github.com/kow3ns/kubernetes-zookeeper]а ...

Вы можете установить spring.kafka.bootstrap-servers = $ {spring.embedded.kafka.brokers} в вашем application.properties для теста, который должен работать. Он заполняется из EmbeddedKafka случайным портом, который был назначен при запуске.

ить FYI:пример работы с gitHub [https://github.com/YunaBraska/embedded-kafka-example] Я искал в интернете и не смог найти работающего и простого примера встроенного теста Кафки. Моя настройка: Весенний ...

Извините - не уверен, куда это поставить - я не парень из потоков; Я просто взломал ваши материалы, чтобы исключить встроенного брокера (так как это в вашем названии вопроса). Рад попробовать для вас, если вы можете быть более откровенным с тем, что вы хотите, чтобы я сделал. Но это будет завтра; поздно.

хожу в поисках знания тайного. Во-первых, у меня есть две пары тем, по одной теме в каждой паре, которая входит в другую тему. Два KTables формируются последними темами, которые используются в KTable + KTable leftJoin. Проблема в том, что ...

Не стесняйтесь принять мой ответ, если он работает сейчас :)

даю данные обработки KTable из KStream. Но когда я запускаю сообщения-захоронения с ключом и нулевой полезной нагрузкой, это не удаляет сообщение из KTable. образец - public KStream<String, GenericRecord> ...