kafkaでpublishのロジックを独自に作ってデータロスを無くす話

class: center, middle

# publishのロジックを独自に作ってデータロスを無くす話

[2017/07/06 Kafka Meetup #3](https://kafka-apache-jp.connpass.com/event/58619)

![CC-BY-NC-SA](https://licensebuttons.net/l/by-nc-sa/3.0/88x31.png)

---
class: middle

- twitter [@xuwei_k](https://twitter.com/xuwei_k)
- github [@xuwei-k](https://github.com/xuwei-k)
- blog <http://xuwei-k.hatenablog.com>

---
class: middle, center

先月(2017年6月)

## Kafka 0.11 リリース🎉

---
class: middle, center

## Exactly Once !?

[Exactly-once Semantics are Possible: Here’s How Kafka Does it](https://www.confluent.io/blog/exactly-once-semantics-are-possible-heres-how-apache-kafka-does-it/)

---
class: middle, center

ひたすら

At Least Once

を死守するためにやったことの話をします・・・

---
class: middle, center

### 1年半前

<https://xuwei-k.github.io/slides/kafka-matsuri>

---
class: middle, center

### 1年ちょっと前

---
class: middle

### スライド一枚で説明

- イベントハブとしてkafka使っている
- Java/Scala クライアントをhttpなどで隠蔽して社内他チームに提供(直接kafka触らせない)
- RAIDコントローラーエラー事件(**中途半端な**死に方して、publish失敗する)
- brokerのversionアップ時にもpublish微妙に失敗する

---
class: middle

### 思いついた and/or 提案してもらった対策

- RAID やめる
- 1つのTopicあたりPartitionを2つ以上作って独自に頑張る (ある意味そういうことしてる、というのを以前のkafka meetupで聞いた気がする)

---
class: middle, center

## "Partition 2つ以上作って独自に頑張る" 案で勝利

---
class: middle

![正常時](kafka-01.png)

---
class: middle

![異常時](kafka-02.png)

---
class: middle, center

基本的な考え方

## partitionを冗長度を上げるため(だけ)に使う

---
class: middle

## やったこと

1. borkerを必ず6台以上用意
2. 1つのtopicにつき必ず2つ以上partitionを用意
3. topic/parition生成時の割合のロジックを自作
4. publish部分を独自に作る

---
class: middle, center

borkerを必ず6台以上用意

## replica 3 ✕ partition 2 = 6台必要

---
class: middle, center

## 1つのtopicにつき必ず2つ以上partitionを用意

1つのpartitionでは、調子悪くなったときの欠損や遅延を防げなかったので、2つ用意

---
class: middle, center

## kafkaのデフォルトの割当ロジックがイケてない !

これをやり始めたのは0.9系だが 0.11.0.0でも変わってなさそう

<https://github.com/apache/kafka/blob/0.11.0.0/core/src/main/scala/kafka/admin/AdminUtils.scala#L137>

---
class: middle

以下のようなコード(Scala)で確かめられる

<pre>
<code class="scala hljs remark-code" style="font-size: 80%;">scalaVersion := "2.12.2"

libraryDependencies += "org.apache.kafka" %% "kafka" % "0.11.0.0"
</code>
</pre>

<pre>
<code class="scala hljs remark-code" style="font-size: 90%;">import kafka.admin._

List.fill(100000)(
 AdminUtils.assignReplicasToBrokers(
 brokerMetadatas = List(
 100, 200, 300, 400, 500, 600 // 適当なborker名
 ).map(kafka.admin.BrokerMetadata(_, None)),
 nPartitions = 2,
 replicationFactor = 3
 )
).distinct.map(
 _.values.foldLeft(List[Int]())(_ ++ _).toSet.toList.sorted
).foreach(println)
</code>
</pre>

---
class: middle

4台か5台にしか分散しない
(brokerが6台より多くても分散しない!?)

---
class: middle

![割当がイケてない](kafka-03.png)

---
class: middle

- topic名のhash値使って、ランダムにはならないようにしつつ、いい感じに分散されるように頑張った
- [scalacheck](https://scalacheck.org/)というproperty based testingライブラリでテスト書いて重点的にチェックした

---
class: middle

### 具体的な実装の話

- [AdminUtils.addPartitions](https://github.com/apache/kafka/blob/0.11.0.0/core/src/main/scala/kafka/admin/AdminUtils.scala#L276) を使わない (必要な部分だけ残しつつ改造)
- 独自に割当を作った後に AdminUtils の [createOrUpdateTopicPartitionAssignmentPathInZK](https://github.com/apache/kafka/blob/0.11.0.0/core/src/main/scala/kafka/admin/AdminUtils.scala#L483) を呼ぶ

---
class: middle

## 独自のpublishロジック

2つのpartitionに対して

- 普段はラウンドロビン
- 片方が調子悪いと判断したら調子良いpartitionだけに送る
- 回復したか？を判断して、良きタイミングでラウンドロビンに戻す
- 全部調子悪そうだったら、(調子良いpartition見つかるまで)逆にラウンドロビン

---
class: middle

## 公式クライアントがつらい

- 公式clientのリトライ機能は、(場合によっては)失敗するとわかっているところに何度も再送してしまって意味がない!?ので、使わない!!!
- リーダーが存在してるborkerが変更されているエラーが返ってきてるのに、またそこにpublishしようとする？？？など
- タイムアウトもちゃんと設定しないと、届く可能性低いのに無駄に待つことになる
- (最新版で改善されているのかどうかは未確認)

---
class: middle

- kafkaにはpartition選択ロジックをカスタマイズするための専用クラスが用意されてる
- <https://github.com/apache/kafka/blob/0.11.0.0/clients/src/main/java/org/apache/kafka/clients/producer/Partitioner.java>
- しかし 使わなかった/使えなかった

---
class: middle, center

結論としては、partitioner使うと

## 失敗したpartitionの 状態管理が厳しい

---
class: middle

もしpartitioner使いつつ、状態管理するなら?

- グローバル変数経由でやりとり？ テストやりづらくなる、辛い
- リフレクション経由で無理やりどうにかする？ それさえ出来るのか不明、出来たとしても辛いのでやりたくない

---
class: middle

- producerの外で独自に状態管理して、指定する場合はsendメソッドの引数でpartitionを明示
- <https://github.com/apache/kafka/blob/0.11.0.0/clients/src/main/java/org/apache/kafka/clients/producer/Producer.java#L76>

---
class: middle

- 想定通り、これによって(2台以上同時に死なない限り)publish失敗がなくなった
- broker側のversion上げる(broker再起動する)ときも、これがないとpublish微妙に失敗すると思うのですが、みなさんどうしてるのか？
- replicaやackの設定が違う？

---
class: middle, center

## おわり

質問タイムなど