Вопросы с меткой [apache-kafka]

Apache Kafka — это распределённая система, предназначенная для обработки потоков данных в режиме реального времени

Фильтрация
Сортировка
Метки
2 голоса
3 ответа
2k показов

Где хранится offset прочитанных сообщений Кафка?

Например, есть 4 консьюмера(потребителя), которые читаю данные из топика. Как Кафка обеспечивает то, что ранее прочитанные сообщения из топика не будут повторно прочитаны другими потребителями или тем ...
Oleg's user avatar
  • 21
2 голоса
2 ответа
1k показов

Подключение go к apache kafka

В целях отладки развернул локально apache kafka с использованием этого прокта. Модифицировал кофигурационный файл следующим образом: version: '2' services: zookeeper: image: wurstmeister/...
anderson's user avatar
  • 417
2 голоса
1 ответ
132 показа

Чтение из кафки с фильтрацией - архитектурная ошибка?

Верно ли утверждение, что читать сообщения из кафки и потом фильтровать его на основе какого-то параметра - плохая идея и это архитектурная ошибка?
saymynamej's user avatar
2 голоса
1 ответ
411 показов

Как формировать названия для топиков Kafka?

Существует ли выработанные практики/стандарты для именования топиков в Kafka? Есть ли "правила хорошего тона" для этого? В документации и имеющихся книжках не смог найти этой информации. ...
Max Raskolnikov's user avatar
2 голоса
1 ответ
932 показа

Kafka tool и не используемые consumers group

У нас в Kafka используется около десятка рабочих consumer group. Однако в Kafka tool 2.0.1 отображаются все consumer(ы) за всю историю ее использования. Хотелось бы убрать лишние, но не знаю как это ...
wedoca's user avatar
  • 381
2 голоса
0 ответов
61 показ

Гарантированная обработка запросов в распределенном приложении

Мне нужна консультация по Java приложению. У меня есть распределенное на кластера приложение A и внешние приложения B, C, D и остальные, всего ~ 50. В мое приложение A идет поток запросов клиентов, ...
mysqllearn's user avatar
2 голоса
1 ответ
40 показов

kafkajs disconnect занимает много времени

Использую либу kafkajs для автотестов. При вызове await consumer.disconnect() - команда выполняется около 5ти секунд, есть какие-то варианты безопасного и быстрого дисконнекта?
Ruslan's user avatar
  • 459
2 голоса
1 ответ
288 показов

Kafka Apache сбрасывает offset

Проблема в следующем: Produser после перезагрузки компа сбрасывает offset и начинает отсчет с 0. т.е. вчера, например, дошел до offset[6589],а сегодня опять offset[0], поэтому consumer не может ...
AriUmu's user avatar
  • 21
2 голоса
0 ответов
52 показа

Как лучше реализовать доступ к одинаковым данным в реальном времени

Добрый день. Есть десяток скриптов, на разных серверах, которые ежеминутно генерируют файл динамичный по содержанию, но примерно одинаковый по структуре. Структура файла не стандартная (не XML, Json),...
Evgeniy A's user avatar
  • 424
1 голос
1 ответ
171 показ

Что из себя представляет значение объекта класса java.util.Date в кафка-сообщении?

Мой сервис отсылает сообщение в формате DTO с разными типами полей - String, Integer, Date итд. Kafka producer сериализует это DTO, kafka consumer в другом сервисе десериализует. Через KafkaTool я ...
gianlucavachhi's user avatar
1 голос
1 ответ
538 показов

Можно ли читать параллельно в нескольких процессах/потоках из одной партиции Kafka?

У меня есть кластер и топики с 3 партициями. По умолчанию когда я забираю батч из кафки, я сдигаю оффсет после прочтения всех сообщений. Могу ли я параллельно читать в нескольких потоках из этой ...
Василий Никпуп's user avatar
1 голос
1 ответ
2k показов

Как передавать объекты большого объема в Kafka

Сразу нужно отметить -- вопрос чисто академический!!! Очевидно, что брокеры сообщений не предназначены для таких вещей. Но это стоит обсудить, чтобы иметь более полное понимание всех возможностей. ...
Vasin Yuriy's user avatar
1 голос
1 ответ
382 показа

Как в kafka определить кто подключен

Добрый день. Прошу подсказать имеется ли какая-то возможность посмотреть в kafka (может в zoookeerer) кто ее использует(подключен или т.п.) Можно конечно смотреть что-то через ZooInspector, но там нет ...
srg321's user avatar
  • 85
1 голос
1 ответ
1k показов

Добавить header в сообщение kafka producer на python

Имеется kafka producer написанный на python и работающий по принципу прокси. То есть на продюсер приходит сообщение извне, обрабатывается и перенаправляется в кафку. Но столкнулся с ошибкой: in send ...
Victor's user avatar
  • 13
1 голос
1 ответ
1k показов

номер порта при запросе на REST API endpoint в микросервисной архитектуре

Я не имею большого опыта в микросервисах и не совсем понимаю как поступить в нижеуказанной ситуации. Есть два вебсервиса, продюсер и консьюмер. Общаются через Apache Kafka, собственно продюсер кидает ...
yakob's user avatar
  • 33
1 голос
1 ответ
493 показа

Организовать очередь сообщений

Есть задача организовать систему передачи сообщений от клиента к менеджеру. Сейчас используется со стороны клиента ejabber , со стороны менеджера "своя система приема и отправки сообщений" . Думаю что ...
user212819's user avatar
1 голос
1 ответ
273 показа

Как узнать дату последнего смещения тем в Kafka?

У меня есть Kafka множеством топиков. И есть Consumer Kafka для чтения записей из этих тем: public Set<ConsumerRecord> consumeKafka() { consumer.subscribe(topics); Set<...
Леонид Дубравский's user avatar
1 голос
1 ответ
357 показов

Зеркальное копирование данных из топика Apache-Kafka

Как при помощи утилиты зеркального копирования kafka-mirror-maker.sh скопировать данные из топика source-кластера Kafka в топик target-кластера с другим именем? К примеру: в первом кластере Kafka ...
Александр Сергеев's user avatar
1 голос
1 ответ
1k показов

org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'kafkaListenerContainerFactory'

Подключил Kafka в Spring Boot, взяв нужные классы из другого проекта, в котором реализован простой Spring, на Tomcat. Зависимость в сборке: compile 'org.springframework.kafka:spring-kafka:2.1.0....
Вячеслав Чернышов's user avatar
1 голос
1 ответ
406 показов

Проблемы с запуском apache-kafka после перезагрузки сервера

Добрый день! Я выставил приоритет загрузки сервисов на одном из серверов. Все сервисы стартуют корректно, за исключением кафки, которая стартует последней. Скидываю лог кафки после запуска сервера. (...
Александр Сергеев's user avatar
1 голос
0 ответов
16 показов

Отправка event при заполнении лога Apache Kafka

Существует приложение на Django, использующее транзакционную обработку потоков. Для исполнения необходимых функций была выбрана Kafka. При заполнении лога необходимо отправлять event, который, в ...
1 голос
0 ответов
36 показов

В чем различия OpenFeign и Kafka?

Руководитель требует разделить данную связь между мкиросервисами на FeignClient и на kafka. В каких случаях лучше использовать FeignClient для взаимодействия между микросервисами, а когда Kafka?
Vladislav Vorobyov's user avatar
1 голос
0 ответов
57 показов

Clickhouse kafka engine. Как задать интервал обращения таблицы к Kafka

Мне необходимо настроить таблицу КХ с движком Кафки так, чтобы таблица брала данные из Кафки большими кусками, но не постоянно, а допустим, с интервалом 30 секунд или 1 минуты. Возможно ли так сделать,...
Андрей's user avatar
1 голос
1 ответ
789 показов

Как получить сообщения из топика Kafka?

Написал типичную функцию для получения и обработки сообщений из топика Kafka. В качестве инструментов использую язык Python и модуль Confluent Kafka. При использовании метода poll() одного из consumer'...
Савелий Зайцев's user avatar
1 голос
0 ответов
121 показ

Clickhouse заходит в кафку по порту отличному от указанного в настройках

В конфиге не указан порт, а только креды на подключение. В таблице clickhous'а указан порт, который нужен для доступа к топику kafka, я там выставил порт 9093. /etc/clickhouse-server/config.xml <...
Marina Vinograd's user avatar
1 голос
0 ответов
31 показ

Как передать в топик Kafka финальный groupBy()

Прошу Вас помочь разобраться с KafkaStreams. Есть код, который делает обычный groupBy(): public static void createGroupByStream(final StreamsBuilder builder) { KStream<String, Person> ...
Falltroy's user avatar
1 голос
0 ответов
25 показов

Удаляются кластеры в kafka tool 2.0.1

Каждое утро при входе в кафка тул, обнаруживаю, что все коннекты удалены. Приходится заново подключаться вручную. Можно как-то решить эту проблему?
Дарья's user avatar
1 голос
0 ответов
312 показов

Сериализация даты как timestamp при записи в кафку

У меня есть json, и в нем есть поле с датой startDt=2023-01-17T08:59:42.522476Z[UTC], когда я кладу json в кафку, то данное поле принимает там значение 1673945982.817186000, нужно чтобы было как в ...
kol9mbla's user avatar
1 голос
0 ответов
70 показов

Fluentbit (как отключить обёртку JSONа в JSON)?

В nginx пишутся логи в JSON формате далее Fluentbit берёт эти логи и отправляет в Kafka, смотрю что прилетает в Kafka и вижу что Fluentbit мой JSON (из nginx) обернул ещё раз в свой JSON. Получился ...
Kirill.ADV's user avatar
1 голос
0 ответов
40 показов

Как можно реализовать хранилище истории изменения сущностей в системе?

Мне нужно реализовать сервис, суть которого в хранение записей о событиях которые произошли в системе. Используется .NET6. События читаются из событийной шины Кафки и записываются в колоночную TSDB, ...
Andrey Nesterov's user avatar
1 голос
1 ответ
895 показов

Spring Kafka - Задержка времени у @KafkaListener при использовании seekToTimestamp у AbstractConsumerSeekAware

Пытаюсь реализовать Rest API с использованием Kafka. POST запрос обрабатывается быстро очень быстро и никаких нареканий к нему нет. GET запрос отрабатывает не совсем корректно и с заметной задержкой ...
Сергей Воронин's user avatar
1 голос
0 ответов
83 показа

apache kafka, как отследить/детекитровать событие - необработанные консюмером сообщения удалены retention политикой

Бывают случаи, когда консюмер по каким-то причинам не обработал сообщения, а retention политика уже эти сообщения удалила, и хотелось бы реагировать на это событие, например переотправить сообщения. ...
user2854511's user avatar
1 голос
1 ответ
649 показов

Как отключить уровень логировния Debug при подключении кафки

Есть вот такой код: fun main() { val props = Properties() props[StreamsConfig.APPLICATION_ID_CONFIG] = "first-stream-app" props[StreamsConfig.BOOTSTRAP_SERVERS_CONFIG] = "127.0.0.1:9092" ...
Василий Никпуп's user avatar
1 голос
0 ответов
53 показа

Как работает Kafka и RabbitMQ через NAT

Доброго времени суток! Я разрабатываю распределенное приложение. Необходимо выбрать систему для обмена сообщениями. Выбор пал между Kafka и RabbitMQ. У меня возник как будет работать мое приложение ...
ruslan's user avatar
  • 116
1 голос
0 ответов
561 показ

AspNetCore + apache Kafka помогите разобраться

С Kafka раньше не работал Здравствуйте, необходимо наладить коммуникацию с Java приложением, решил попробовать Kafka как посредника. Работаю под Win10. Клиент для kafka от confluence для AspNetCore. ...
Aldmi's user avatar
  • 1,925
1 голос
1 ответ
330 показов

Не создается topic KAfka

Всем привет. У меня есть приложение, которое работает с Stream Kafka. Я в нем также пытаюсь создать топики но получаю ошибку: INFO Terminate ZkClient event thread. (org.I0Itec.zkclient....
Леонид Дубравский's user avatar
1 голос
0 ответов
131 показ

Apache kaffka запуск тестового consumer приводит к ошибке "Java.lang.OutOfMemoryError"

Запускаю Apache kaffka на компьютере с Win7 с помощью стандартных команд (указанных в QuickStart) Zookeeper START. E:\kafka_2.11-2.0.0\bin\windows\zookeeper-server-start.bat E:\kafka_2.11-2.0.0\...
Aldmi's user avatar
  • 1,925
1 голос
1 ответ
683 показа

Multitheading spring kafka consumer

Есть топик в kafka c одной партицией. Обработка одного сообщения из топика занимает продолжительное время. Можно ли написать многопоточный spring kafka consumer, чтобы успевать обрабатывать все данные,...
Aisha's user avatar
  • 31
0 голосов
2 ответа
635 показов

Не могу подключиться к kafka из docker контейнера

Запускаю сервисы, а также kafka и zookeeper в контейнерах докер. Запускаю сервис akhq , который являются как бы панелью для просмотра топиков и сообщений кафка. В сервисе akhq отображаются созданные ...
Андрей Алексеев's user avatar
0 голосов
1 ответ
191 показ

Нормально ли в Kafka создать множество топиков для общения с сервисами?

У меня есть микросервисы и некий оркестратор, который управляет их работой. Пытаюсь понять нормально ли поступать следующим образом: К нам пришло сообщение на оркестратор через топик 1. Оркестратор ...
Василий Никпуп's user avatar
0 голосов
1 ответ
25 показов

Какие подходы используете для иницилизации "нулевого" состояния для нового микросервиса в экосистеме Kafka?

Стандартная EDA архитектура с Kafka. События удаляются через какое-то время (пусть будет неделя). Собственно, каждый микросервис по получении событий собирает копию тех данных, которые ему нужны от ...
Egor Kozelskij's user avatar
0 голосов
1 ответ
67 показов

JDBC Connector Confluent Kafka

Можно ли в конфигурации JDBC sink Connector Kafka каким-нибудь образом вызвать процедуру? Ознакомился с https://docs.confluent.io/kafka-connect-jdbc/current/sink-connector/sink_config_options.html Там ...
Геннадий Михайлов's user avatar
0 голосов
2 ответа
407 показов

Как подключиться spring микросервису из docker контейнера к kafka в контейнере?

Создаю контейнеры через docker-compose up version: '3' services: zookeeper: image: confluentinc/cp-zookeeper:latest environment: ZOOKEEPER_CLIENT_PORT: 2181 ZOOKEEPER_TICK_TIME: ...
Sergey M's user avatar
0 голосов
2 ответа
6k показов

Запустить kafka в docker, и получить к ней доступ из внешнего сервиса

Только начал изучать докер. Идея следующая: Есть 2 spring boot сервиса, между которыми общение осуществляется через кафку. Хочу попробовать перенести все это в докер. Для начала решил перенести в ...
Arty Mart's user avatar
  • 154
0 голосов
1 ответ
565 показов

Как после consumer rebalance процесса, сбросить offset, что бы Consumer мог прочитать топик сначала?

Есть ситуация, когда Consumer1 читает сообщения из кафка топика. При подключении второго Consumer2 с таким же groupId, происходит перераспределение партиций, можно ли как то сбросить offset, что бы ...
drugzverya's user avatar
0 голосов
1 ответ
566 показов

как указывать в Кафка какому микросервису адресовано сообщение?

я новичок в Кафка. Теории изучил много, но на практике уже не понимаю как лучше с кафкой работать. как указывать в Кафка какому микросервису адресовано сообщение, если все микросервисы читают из общей ...
Андрей Алексеев's user avatar
0 голосов
1 ответ
496 показов

Почему при отправке сообщений в Кафке, если не правильно написан топик которого не существует то сообщения все равно уходят и нет ошибки? [закрыт]

Почему при отправке сообщений в Кафке, если не правильно написан топик которого не существует то сообщения все равно уходят и нет ошибки?
works-mails's user avatar
0 голосов
1 ответ
178 показов

Как правильно распределять запросы по микросервисам?

Допустим в качестве брокера запросов к микросервисам использую Кафка или Rabbit или что то другое. Пока выбираю что использовать. Но я не понимаю на основании чего брокеры знают каким именно сервисам ...
Андрей Алексеев's user avatar
0 голосов
1 ответ
32 показа

Flink CEP и parallelism

Произвожу обработку входящего потока с кафки при помощи flink cep. Описал pattern. Указал параллелизм у StreamExecutionEnviroment. Во flink dashboard вижу цепочку, где сама обработка сообщения с ...
Konstantin's user avatar
0 голосов
1 ответ
365 показов

Как вычитать с помощью спарка полный топик из кафки в batch mode

В гугле на тему spark/kafka попадаются примеры со стриммингом, как я могу в batch моде вычитать данные из кафки используя spark(любое api java/scala).
Андрей Козицкий's user avatar