Сочный DevOps on Telegram: “Фиксим partition skew и leader skew в kafka Мы д…”

Фиксим partition skew и leader skew в kafka

Мы для удобства повседневной работы с kafka кластерами используем kafka-ui.
Увеличивали RF на топике, увидели что partitions skew у одного брокера и leader skew у другого слегка стали брать больше нагрузки чем другие.

Смотрим распределение лидеров партиций у брокеров (слева кол-во партиций, справа broker id):

/usr/local/kafka/bin/kafka-topics.sh --bootstrap-server $BS --command-config $CFG --describe \ | awk '{for(i=1;i<=NF;i++) if ($i=="Leader:") print $(i+1)}' \ | sort | uniq -c | sort -nr

61 2 
58 1 
53 4 
52 3 
51 5 
45 6

Сильнее всех отстает 6 брокер. Второй наиболее загружен.
Смотрим распределение реплик по брокерам:

/usr/local/kafka/bin/kafka-topics.sh --bootstrap-server $BS --command-config $CFG --describe \ | awk ' match($0,/Replicas: \[?([0-9, ]+)\]?/,m){ s=m[1]; gsub(/[ \[\]]/,"",s); n=split(s,a,","); for(i=1;i<=n;i++) cnt[a[i]]++ } END{for(b in cnt) printf "%6d %s\n", cnt[b], b} ' | sort -nr 

128 1 
116 2 
113 5 
106 4 
106 3
99 6

Тут картина немного другая. 1-й брокер имеет наибольшее кол-во реплик. 6-ой - наименьшее.
Возьмем за основу, что 1 брокер - нагруженный, 6 нет.

Генерируем топики для reassign-а:

HEAVY=1
LIGHT=6
/usr/local/kafka/bin/kafka-topics.sh --bootstrap-server $BS --command-config $CFG --describe \
| awk -v H=$HEAVY -v L=$LIGHT '
  function has(b, s) { return index("," s ",", "," b ",") }
  /Topic: / { if (match($0,/Topic: ([^ ]+)/,t)) topic=t[1] }
  match($0,/Replicas: \[?([0-9, ]+)\]?/,m) {
    s=m[1]; gsub(/[ \[\] ]/,"",s)
    if (has(H,s) && !has(L,s)) hit[topic]++
  }
  END{ for (k in hit) printf "%6d %s\n", hit[k], k }' \
| sort -nr \
| grep -v '^__' \
| head -n 10 \
| awk '{print "{\"topic\":\""$2"\"}"}' \
| paste -sd, - \
| sed '1s/^/{"topics":[/; $s/$/], "version":1}/' \
> /tmp/topics_to_move.json

Выполняем так называемый reassign. Сначала генерируем план для него на основе наших топиков.

/usr/local/kafka/bin/kafka-reassign-partitions.sh \
  --bootstrap-server $BS --command-config $CFG \
  --topics-to-move-json-file /tmp/topics_to_move.json \
  --broker-list "1,2,3,4,5,6" \
  --generate | tee /tmp/reassign.gen

Далее достаем JSON из вывода:

awk '/Proposed partition reassignment configuration/{p=1} p{print} /\}$/ {if(p){exit}}' /tmp/reassign.gen \
| sed -n '/{/,/}/p' > /tmp/reassign-plan.json

Выполняем сам reassign с тротлингом 50 МБ/с:

 /usr/local/kafka/bin/kafka-reassign-partitions.sh \
  --bootstrap-server $BS --command-config $CFG \
  --reassignment-json-file /tmp/reassign-plan.json \
  --throttle 52428800 \
  --execute

Последить за выполнением можно так:

watch -n 5 "/usr/local/kafka/bin/kafka-reassign-partitions.sh \
  --bootstrap-server $BS --command-config $CFG \
  --reassignment-json-file /tmp/reassign-plan.json --verify"

Ну и в UI ждем, когда все реплики за синкаются.
Если вдруг лидеры не выровнялись по завершению, то можно попробовать:

/usr/local/kafka/bin/kafka-leader-election.sh \
  --bootstrap-server $BS --command-config $CFG \
  --election-type PREFERRED \
  --all-topic-partitions

#kafka