TT Lab
はじめる
学ぶ 学習パス コース

注文が二度届き、一度は消えた

保持とコンパクション — トピック設定が残すもの、捨てるもの

TT Labで続きを見る

このラボはVM上で動きます

UbuntuのVMに、Apache Kafka 4.3.1がKRaftの単一ノードとして起動しています (localhost:9092)。ブローカーの保持の検査周期(log.retention.check.interval.ms)を 10秒、クリーナーの待機(log.cleaner.backoff.ms)を5秒に縮めてあるので、1分以内に削除と 圧縮を確認できます。最初の起動には4分ほどかかります。

目標

時間による保持(retention.ms)が古いセグメントを削除すること、圧縮(cleanup.policy=compact)が キーごとに最後の値だけを残すこと、max.message.bytesが大きなレコードを拒否することを、 自分の目で確認して、実行中のトピックの設定を変更してみます。

なぜ重要なのか

トピックの設定は、「何をどれだけ長く残すか」を決めます。デフォルトの保持は7日で、 セグメント単位で削除されるので、retention.msが過ぎたからといって、すぐには消えません。 アクティブなセグメントは決して削除されず、セグメントがロールされて(segment.ms)初めて 削除できます。圧縮は、別の種類の保持です。キーのある変更ログで、各キーの 最後の値を必ず残してくれるので、最初から読めば現在の状態が復元されます。この 違いを知らないと、「削除したのに残っている」「残すと言ったのに消えた」という報告を読み解けません。

ステップ

  1. eventsトピックを、retention.ms=20000、segment.ms=10000で作成してください。
  2. e1・e2・e3を入れて、12秒以上待った後、e4を入れてセグメントをロールさせ、さらに30秒ほど待ってください。一番前のオフセットが0ではなくなったら、/root/kafka/retention.txtファイルに、earliest=<가장 앞 오프셋>、latest=<로그 끝 오프셋>の2行を書いてください(プレースホルダーは、一番前のオフセットと、ログ末尾のオフセットです)。
  3. customersトピックを、cleanup.policy=compact、segment.ms=10000、min.cleanable.dirty.ratio=0.01で作成してください。
  4. キーのパースで、alice:v1、bob:v1、alice:v2、alice:v3を入れて、12秒以上待った後、carol:v1を入れてセグメントをロールさせてください。圧縮が終わったら(30秒ほど)、最初からキーと一緒に読んで、/root/kafka/compacted.txtファイルに保存してください。aliceは、v3が1回だけなければなりません。
  5. photosトピックを、max.message.bytes=1024で作成して、2,000文字の1行(/root/kafka/big2k.txt)を入れてみてください。プロデューサーのstderrを、/root/kafka/toolarge.txtファイルに残してください。拒否されて、レコードは0件でなければなりません。
  6. kafka-configs.shで、eventsのretention.msを86400000に変更してください(トピックは作り直さないでください)。
  7. /root/kafka/topic-report.txtファイルに、events_retention_ms=86400000、customers_policy=compact、photos_max_bytes=1024、min_insync_default=1、three_broker_min_isr=2の5行を書いてください。

参考

20秒だけ残すトピック

eventsトピックを、retention.ms=20000、segment.ms=10000で作成してください。

--create --topic events --partitions 1 --config retention.ms=20000 --config segment.ms=10000。segment.msがないと、デフォルトの7日間はセグメントがロールされず、何も削除されません。

古いセグメントが削除される

e1・e2・e3を入れて、12秒以上待った後、e4を入れてセグメントをロールさせ、さらに30秒ほど待ってください。一番前のオフセットが0ではなくなったら、/root/kafka/retention.txtファイルに、earliest=<가장 앞 오프셋>、latest=<로그 끝 오프셋>の2行を書いてください(プレースホルダーは、一番前のオフセットと、ログ末尾のオフセットです)。

kafka-get-offsets.sh --topic events --time earliestが、events:0:3のように変われば、最初のセグメントが削除されたのです。削除される単位はレコードではなくセグメントなので、e1–e3が一度に消え、アクティブなセグメントのe4は残ります。

キーごとに最後の値だけを残すトピック

customersトピックを、cleanup.policy=compact、segment.ms=10000、min.cleanable.dirty.ratio=0.01で作成してください。

--config cleanup.policy=compact --config segment.ms=10000 --config min.cleanable.dirty.ratio=0.01。dirty ratioのデフォルト値は0.5なので、ログの半分が重複していなければ、クリーナーが動きません。ラボでは、ほぼ0に下げます。

圧縮の後は、aliceが1つになる

キーのパースで、alice:v1、bob:v1、alice:v2、alice:v3を入れて、12秒以上待った後、carol:v1を入れてセグメントをロールさせてください。圧縮が終わったら(30秒ほど)、最初からキーと一緒に読んで、/root/kafka/compacted.txtファイルに保存してください。aliceは、v3が1回だけなければなりません。

アクティブなセグメントは圧縮されないので、carolでロールさせる必要があります。読むときに--formatter-property print.key=trueを指定すると、alice\tv3の形になります。aliceがまだ3つなら、クリーナーが動く前なので、10秒後にもう一度読んでください。

大きすぎるレコードは拒否される

photosトピックを、max.message.bytes=1024で作成して、2,000文字の1行(/root/kafka/big2k.txt)を入れてみてください。プロデューサーのstderrを、/root/kafka/toolarge.txtファイルに残してください。拒否されて、レコードは0件でなければなりません。

head -c 2000 /dev/zero | tr '\0' y > big2k.txt; echo >> big2k.txt。ブローカーがバッチサイズの上限で拒否すると、プロデューサーのログにRecordTooLarge系のエラーが出力され、kafka-get-offsets.shは0を返します。

実行中に設定を変更する

kafka-configs.shで、eventsのretention.msを86400000に変更してください(トピックは作り直さないでください)。

kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name events --alter --add-config retention.ms=86400000。運用ドキュメントの「Modifying topics」のとおりです。--describeで確認してください。

設定値でまとめる

/root/kafka/topic-report.txtファイルに、events_retention_ms=86400000、customers_policy=compact、photos_max_bytes=1024、min_insync_default=1、three_broker_min_isr=2の5行を書いてください。

最初の3つは、kafka-configs.sh --describeで今の値を読んで書きます(採点ツールもそのように見ます)。後ろの2つは、トピック設定のドキュメントのmin.insync.replicasの項目です。デフォルト値と、レプリケーションファクターが3のときの典型的な値です。