VG
Size: a a a
2020 September 15
N
Producer. да
спасибо.у меня тормозит запись при acks = 1 получется где-то 65MB/sec, а при -1 падает до 10MB-15MB. тестирую через стандартные утилиты выставляя batch.size = 512K and ISR = 2 т.е он запишет на лидера и еще на одного фоловера. ну если бы падала в 2 раза запись. былобы обьяснимо, но она больше падает
VG
Nikolay
спасибо.у меня тормозит запись при acks = 1 получется где-то 65MB/sec, а при -1 падает до 10MB-15MB. тестирую через стандартные утилиты выставляя batch.size = 512K and ISR = 2 т.е он запишет на лидера и еще на одного фоловера. ну если бы падала в 2 раза запись. былобы обьяснимо, но она больше падает
А сколько нод?
N
а если бы было другое число нод, а не 4. это могло как-то обьяснять?
N
Nikolay
спасибо.у меня тормозит запись при acks = 1 получется где-то 65MB/sec, а при -1 падает до 10MB-15MB. тестирую через стандартные утилиты выставляя batch.size = 512K and ISR = 2 т.е он запишет на лидера и еще на одного фоловера. ну если бы падала в 2 раза запись. былобы обьяснимо, но она больше падает
напомните -1 это про дождаться ответ от всех?
N
кому интересно. в процессе копания этого ишью обнаружил, что если размер сообщения очень маленький ( у меня сначала было для теста 10 байт, а сейчас сделал 200 байт сообщений), то это сильно просаживает перфоманс отправителя. т.к на каждый send создается Future и он потом все эти Future должен закомплитить и в каждую из них проставить еще offset. Т.е если даже ставить acks = 0, то он все равно потом много времени тратить на все эти миллионы Future
N
а как смотрят на производительность брокеров? как узнать куда он время тратит
PD
Nikolay
кому интересно. в процессе копания этого ишью обнаружил, что если размер сообщения очень маленький ( у меня сначала было для теста 10 байт, а сейчас сделал 200 байт сообщений), то это сильно просаживает перфоманс отправителя. т.к на каждый send создается Future и он потом все эти Future должен закомплитить и в каждую из них проставить еще offset. Т.е если даже ставить acks = 0, то он все равно потом много времени тратить на все эти миллионы Future
Ну так обычно MPS важнее, чем MB/s. Не файлы же передаем.
N
вот смотрю код, который обрабатывает request от продьюсера. может кто копал и может вот этот кусок пояснить
N
if (delayedProduceRequestRequired(requiredAcks, entriesPerPartition, localProduceResults)) {
// create delayed produce operation
val produceMetadata = ProduceMetadata(requiredAcks, produceStatus)
val delayedProduce = new DelayedProduce(timeout, produceMetadata, this, responseCallback, delayedProduceLock)
// create a list of (topic, partition) pairs to use as keys for this delayed produce operation
val producerRequestKeys = entriesPerPartition.keys.map(TopicPartitionOperationKey(_)).toSeq
// try to complete the request immediately, otherwise put it into the purgatory
// this is because while the delayed produce operation is being created, new
// requests may arrive and hence make this operation completable.
delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys)
} else {
// we can respond immediately
val produceResponseStatus = produceStatus.map { case (k, status) => k -> status.responseStatus }
responseCallback(produceResponseStatus)
}
// create delayed produce operation
val produceMetadata = ProduceMetadata(requiredAcks, produceStatus)
val delayedProduce = new DelayedProduce(timeout, produceMetadata, this, responseCallback, delayedProduceLock)
// create a list of (topic, partition) pairs to use as keys for this delayed produce operation
val producerRequestKeys = entriesPerPartition.keys.map(TopicPartitionOperationKey(_)).toSeq
// try to complete the request immediately, otherwise put it into the purgatory
// this is because while the delayed produce operation is being created, new
// requests may arrive and hence make this operation completable.
delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys)
} else {
// we can respond immediately
val produceResponseStatus = produceStatus.map { case (k, status) => k -> status.responseStatus }
responseCallback(produceResponseStatus)
}
N
покритикуйте. вроде понял почему сильно деградирует при смене с 1 на -1. у нас же сначала пишется в лог лидера + потом реплику пулят через определенный интервал. вот на этот интервал всегда будет увеличение для каждой пачки. ну или на половину этого интервала + на время записи в page cache лидера
2020 September 16
N
весь лог брокера забит сообщениями, что InvalidRecordException для __consumer_offsets. Что делать?
Y
скинь полный трейс
НИ
if (delayedProduceRequestRequired(requiredAcks, entriesPerPartition, localProduceResults)) {
// create delayed produce operation
val produceMetadata = ProduceMetadata(requiredAcks, produceStatus)
val delayedProduce = new DelayedProduce(timeout, produceMetadata, this, responseCallback, delayedProduceLock)
// create a list of (topic, partition) pairs to use as keys for this delayed produce operation
val producerRequestKeys = entriesPerPartition.keys.map(TopicPartitionOperationKey(_)).toSeq
// try to complete the request immediately, otherwise put it into the purgatory
// this is because while the delayed produce operation is being created, new
// requests may arrive and hence make this operation completable.
delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys)
} else {
// we can respond immediately
val produceResponseStatus = produceStatus.map { case (k, status) => k -> status.responseStatus }
responseCallback(produceResponseStatus)
}N
Yuriy
скинь полный трейс
В логе нет трейса. Просто вот огромное количество таких сообщений. Это на 2х из 4х брокеров