В смысле?

брокер читает с лидера  в тот момент, когда на лидере нет новых данных для брокера. если это просто polling, то ответ вернется сразу - данных нет, а если long polling то ответ вернется, когда появятся данные ( или будет превышен интервал ожидания)

так понимаю, что с брокера просто pooling.

override def buildFetch(partitionMap: Map[TopicPartition, PartitionFetchState]): ResultWithPartitions[Option[FetchRequest.Builder]] = {
     val fetchData = mutable.Map.empty[TopicPartition, FetchRequest.PartitionData]
     partitionMap.foreach { case (partition, state) =>
       if (state.isReadyForFetch) {
         val replicaState = replicaPartitionState(partition)
         fetchData.put(partition, new FetchRequest.PartitionData(state.fetchOffset, replicaState.logStartOffset,
           1024 * 1024, Optional.of[Integer](state.currentLeaderEpoch)))
       }
     }
     val fetchRequest = FetchRequest.Builder.forReplica(ApiKeys.FETCH.latestVersion, replicaId, 0, 1, fetchData.asJava)
     ResultWithPartitions(Some(fetchRequest), Set.empty)
   }

вот тут val fetchRequest = FetchRequest.Builder.forReplica(ApiKeys.FETCH.latestVersion, replicaId, 0, 1, fetchData.asJava)

не. ведь он ждет до 1го байта. так ведь? значит long polling?

хотя нет . вот нашел вроде  // respond immediately if 1) fetch request does not want to wait
   //                        2) fetch request does not require any data
   //                        3) has enough data to respond
   //                        4) some error happens while reading data
   //                        5) any of the requested partitions need HW update
   if (timeout <= 0 || fetchInfos.isEmpty || bytesReadable >= fetchMinBytes || errorReadingData || anyPartitionsNeedHwUpdate)

тогда что выходит, что реплика постоянно бомбит лидера запросами без всяких ожиданий?

опа. не туда смотрел. тот код с нулем был из теста. если кто читал сообщения, то пардон. там или 1 байт или replica.fetch.wait.max.ms и вот документ  https://cwiki.apache.org/confluence/display/KAFKA/Purgatory+Redesign+Proposal   "A fetch request with min.bytes=1 won't be answered until there is at least one new byte of data for the consumer to consume. This allows a "long poll" so that the consumer need not busy wait checking for new data to arrive"

в телеграме можно без пардонов и мерси удалять сообщения, прогресс уже там

Мы параллельно обрабатываем сообщения вычитанные из партиций кафки в очередь. Обработка так гораздо быстрее чем последовательно. Но при этом для коммита надо складывать обработанные оффсеты в непрерывную последовательность и потом коммитить иначе при восстановлении в случае сбоя не обработанные сообщения могут быть потеряны. Также приходится очень тщательно обрабатывать ошибки и следить за тем что код занимающийся параллельной обработкой не должен зависать. Если одно сообщение зависнет вас может порвать по памяти при накоплении непрерывной последовательности. Готового решения не встречал. Но в доке кафки написано что если захочется параллельной обработки партиции то надо делать примерно так как я описал.

хорошо, спасибо, тогда наверно выложу свою поделку, может кто посмотрит скажет где не прав

в коннекторы параметры внутренних консюмеров не прокинуты, к сожалению

PR мой год уже висит, @gamussa доколе? :)

а backpressure не делаете, чтобы зависание не роняло приложение?

Можно ведь переопределить настройки клиентов https://docs.confluent.io/current/connect/references/allconfigs.html#override-the-worker-configuration
или это про другое?

наверно уже вынесли, раньше не было

у нас очереди на каждую партицию отдельные и ограниченные
если очередь заполнена перестаем читать партицию пока не разгребется очередь

зависание кода обработчика мы не можем контролировать (там пользовательский код)
поэтому у нас есть таймаут и если в течении этого таймаута код обработчика не возвращает управление то мы просто добавляем ооффсет этого сообщения к последовательности на коммит

тут выбор либо блокировать и выходить изз групппы подписки либо продолжать с подвисшим потоком, мы выбрали продолжать

Всем привет)
Кто-то сталкивался с проблемой, что Оффсет не проставляется?

Подробнее)

Парни, может вопрос нуббский