KafkaのExactly-OnceでML推論の重複が消える＋信頼性も上がるよ
- ML推論リクエスト1件ごとにキーを決めて、24時間以内に同じキーが2回入ったら即エラー返す仕組みを作ると、重複実行をほぼゼロにできるよ。
KafkaのExactly-Once処理と安定キーを組み合わせると、ダブり推論の検出精度が99%超える（3日間テストで同一キー重複0件ならOK）。
- 推論開始から10分以内にトランザクションで書き込み→コミットまで完了させて、途中エラー時は強制ロールバックしよう。
10分以内で一連の処理を終わらせることで、中断時のゴミデータ流入や推論漏れリスクがグッと減る（再実行時の出力ダブり率5%以下で確認）。
- Kafka ConsumerでPoll→処理→書き込み→コミットまで一気通貫（各100ms以内）で流すと、リアルタイム性能とExactly-Once両立できる。
100ms単位で処理を流すと、レイテンシ気にならないし、重複出力も抑えやすい（1時間あたり遅延0.5秒以内＆重複0件なら成功判定）。
- 副作用あるAPI呼び出しは全部ログ保存してから発火！30分以内に失敗分だけ自動再送する仕組みにすると信頼性が一気に増す。
ログで呼び出し管理すると、同一リクエストの二重実行が激減（1週間運用で手動リカバリー発生0件なら十分）。
なぜKafkaのExactly-Once処理がML推論で重要なのか知ろう
・ねえ、リアルタイムMLやるなら「exactly-once」運用、ガチで最重要ポイント！！！同じイベントに対して答えが2個出てきたら？終わりでしょ。ていうか、違う結果混ざった瞬間に一気に信頼消滅するんだよなー…。
・「exactly-once」はただのチェックリストじゃない！むしろ考え方そのもの。Kafka使うときって、トピック・キーの設計からプロデューサー冪等性、トランザクション管理、あとRead-Process-Writeパターンの作り込み…うっかり全部関係してくるんだわ！
・それからね、適当にリトライ入れたり雑な実装してると出力がダブる可能性バリ高だから、重複防止仕組みは絶対セットで欲しい！！！いや、本音いうと一番タチ悪いのは「同じ予測結果二度返し、ごめん！」みたいなパターン…これコスパ最悪。さらに監査や追跡まで必要なら余計にconsistency意識ないとマジ詰む感じかもしれない。
・ちなみにさ、トレーニングはバッチ処理だと何回でもやり直せるけど、本番推論中はそれ無理っぽい。現場だと本当に1発勝負みたいになるよ！
・だからKafka組む時は絶対「exactly-once脳」で動線や接続の全設計見直したほうがいいってば！結果的に運用コスト抑えられるし、不具合もサクッと見つかる確率上がると思う！！わかるかな？
・「exactly-once」はただのチェックリストじゃない！むしろ考え方そのもの。Kafka使うときって、トピック・キーの設計からプロデューサー冪等性、トランザクション管理、あとRead-Process-Writeパターンの作り込み…うっかり全部関係してくるんだわ！
・それからね、適当にリトライ入れたり雑な実装してると出力がダブる可能性バリ高だから、重複防止仕組みは絶対セットで欲しい！！！いや、本音いうと一番タチ悪いのは「同じ予測結果二度返し、ごめん！」みたいなパターン…これコスパ最悪。さらに監査や追跡まで必要なら余計にconsistency意識ないとマジ詰む感じかもしれない。
・ちなみにさ、トレーニングはバッチ処理だと何回でもやり直せるけど、本番推論中はそれ無理っぽい。現場だと本当に1発勝負みたいになるよ！
・だからKafka組む時は絶対「exactly-once脳」で動線や接続の全設計見直したほうがいいってば！結果的に運用コスト抑えられるし、不具合もサクッと見つかる確率上がると思う！！わかるかな？
リアルタイムML予測の信頼性を保つ方法を見つけよう
Inferenceって、なんかそれだけやっとけばいいって思いがちだけど、正直言うと全然足りないんだわ…。例えばさ、支払いリクエストがFeature Storeとかモデル宛てに投げられる時に、それが同じ内容で2回処理されたら…いや〜地味に最悪。下手するとユーザーに二重請求しちゃうし、同じメールも何回も届いたりするし、多分だけどデータ分析の結果までぐちゃぐちゃになりそう。
Kafka使うなら、その冗長プロデューサー機能？あとトランザクションね。こういうの駆使するとさ、「本当にexactly-onceやるぞ！」みたいな土台は作れるっぽいよ。イメージ的には、「各inputイベントにつき一発で耐久性ありつつ追跡もできる結果出す」みたいな…うーん、この辺がポイントなんだよな。でも結局は「そのイベント自体をちゃんと状態判定の根拠にする」って考え方じゃないとダメってこと。
もし同じinputキーがもう1回来た場合は、とりあえず二択だと思ってて。「まったく同じ決定論的なレス返す」か、「前も出したやつ証明できれば何もしない」って、そのどっちか選ぶべき。…他は特になくてさ。
正直デカいフレームワークとかいらんよ。ほんとに必要なのは、厳選した制約を数個入れておけばまぁ十分かな、みたいな感覚だわ。
Kafka使うなら、その冗長プロデューサー機能？あとトランザクションね。こういうの駆使するとさ、「本当にexactly-onceやるぞ！」みたいな土台は作れるっぽいよ。イメージ的には、「各inputイベントにつき一発で耐久性ありつつ追跡もできる結果出す」みたいな…うーん、この辺がポイントなんだよな。でも結局は「そのイベント自体をちゃんと状態判定の根拠にする」って考え方じゃないとダメってこと。
もし同じinputキーがもう1回来た場合は、とりあえず二択だと思ってて。「まったく同じ決定論的なレス返す」か、「前も出したやつ証明できれば何もしない」って、そのどっちか選ぶべき。…他は特になくてさ。
正直デカいフレームワークとかいらんよ。ほんとに必要なのは、厳選した制約を数個入れておけばまぁ十分かな、みたいな感覚だわ。
安定したキーとスキーマ設計で重複防止を実現しよう
1. Kafkaのリクエストトピック作る時、「order_id」とか「payment_id」みたいなビジネス用のキーをKafka keyにしておく方が無難。何でもいいやって適当なユニークID振るのは、まあ楽なんだけど…後で後悔することあるっぽい。あんま考えずにやると困るパターン多いかな、うん。
2. ずっとJSONだけ使い続けたい人いるよね。でもさ、バージョン付きスキーマ（例：AvroやProtobuf）で運用した方が確実にいい。理由？リプレイとか遅延イベントが来てもデータ構造が揃ってて処理しやすいし、一貫して確認できる。そこはガチ。
3. idempotent producer有効化する？したほうがいいかも。あとトランザクション機能つけた方が安全。一度にまとめて書き込む＆オフセットコミットも同時に処理できるから、全部失敗か全部成功。中途半端な状態になりづらい。わりと大事っす。
4. あとConsumer側、isolation.level=read_committedで読む設定しないとダメだよ。「これ忘れると未完成の書き込み結果を普通に見ちゃうケースある」…マジ注意ね、ホント。
5. モデルコードから直接サードパーティAPI呼び出すの禁止！ログ残さない副作用系処理もダメ、基本NG。「実行前に絶対ちゃんとログ取っとけ」ってこと、まあそういう話ですわ。
2. ずっとJSONだけ使い続けたい人いるよね。でもさ、バージョン付きスキーマ（例：AvroやProtobuf）で運用した方が確実にいい。理由？リプレイとか遅延イベントが来てもデータ構造が揃ってて処理しやすいし、一貫して確認できる。そこはガチ。
3. idempotent producer有効化する？したほうがいいかも。あとトランザクション機能つけた方が安全。一度にまとめて書き込む＆オフセットコミットも同時に処理できるから、全部失敗か全部成功。中途半端な状態になりづらい。わりと大事っす。
4. あとConsumer側、isolation.level=read_committedで読む設定しないとダメだよ。「これ忘れると未完成の書き込み結果を普通に見ちゃうケースある」…マジ注意ね、ホント。
5. モデルコードから直接サードパーティAPI呼び出すの禁止！ログ残さない副作用系処理もダメ、基本NG。「実行前に絶対ちゃんとログ取っとけ」ってこと、まあそういう話ですわ。
トランザクションと冪等性を活用して出力の一貫性を維持しよう
おお！Kafkaにアウトボックスレコード投げる！勢いでバーンって送信するやつね。すぐ後ろでコネクターとか小さいWorkerが待機してて、そいつらがサイドエフェクト処理をガシッと適用しちゃう構成。超動的。あっ、それと、もし再試行発生しても新規イベント作り直さないんだよ。同じイベントをもう一回（冪等だよ！）リプレイする方式だから安心…たぶんね！
「exactly-once thinking」の頭の中、めっちゃシンプルな4ステップなんだわ。Read→Decide→Write→Commit。この流れ一個でもコケたら全部ロールバックで最初からフルリセット！！無駄な意思決定プロセス省けるしスッキリよ。
まずRead。ビジネスID単位のキー付いたレコードをインプットトピックのパーティションからピックアップするだけ！短い話。それからDecideへ突入。そのキー特徴量引っ張ってきたり、モデル走らせたり結構自由。そして結果には必ず入力のキー＋deterministicなdecision_id（例:なんかハッシュとか）ぶち込むスタイルになる！速攻ね～。
「exactly-once thinking」の頭の中、めっちゃシンプルな4ステップなんだわ。Read→Decide→Write→Commit。この流れ一個でもコケたら全部ロールバックで最初からフルリセット！！無駄な意思決定プロセス省けるしスッキリよ。
まずRead。ビジネスID単位のキー付いたレコードをインプットトピックのパーティションからピックアップするだけ！短い話。それからDecideへ突入。そのキー特徴量引っ張ってきたり、モデル走らせたり結構自由。そして結果には必ず入力のキー＋deterministicなdecision_id（例:なんかハッシュとか）ぶち込むスタイルになる！速攻ね～。
ログによる副作用制御で外部APIとの信頼性を高めよう
うわ、KafkaのJava運用、このread-process-writeサイクル、マジで説明したくなるやつ！いきなりだけど、「inference.requests」トピックからゴソッとメッセージ拾って、それをガンガン「inference.decisions」に流すんだ。その上でオフセットを一気にまとめてコミットする設計。エモすぎない？いや普通にスカっとしない？
それでね、Gradleには例の「org.apache.kafka:kafka-clients:3.7.0」これ記載が必須！プロデューサ側はちょっと細かく攻めるのがキモ。「bootstrap.servers」は"kafka:9092"。んで、「enable.idempotence」、忘れたら後悔する設定。あと「acks」はall設定確定。それから…ええと、「max.in.flight.requests.per.connection」が5固定、「retries」にはInteger.MAX_VALUEド直球突っ込む。でさ「transactional.id」を"inference-tx-1"みたいに毎Producerごと別々！初期化はもちろんproducer.initTransactions()無視禁止。
次コンシューマーね？そう、「kafka:9092」で繋いでOK。「group.id」は"inference-workers"に固定しちゃって平気。「enable.auto.commit」はfalse一択だわ。そして激推しなのが「isolation.level」を"read_committed"指定、ここほんと半端ダメ絶対ポイントだから！中途半端な未コミットデータとか怖すぎ。
で、本番ループ話いくよ？まずpoll(500ms)でバッチ吸い上げ→空なら即戻り。でもレコード入ってきたらproducer.beginTransaction()→爆速トランザクション開始。そのままループ回してkey（例えばpayment_id）もpayloadもゴソッて抜き出してrunModel(payload, "fraud-v42")実行。このrunModelさぁ、payload×model_version=判定結果、要するに完全な決定論型だと思う。
出てきたInferenceDecisionはserialize（直列化）してから新しいProducerRecord化→どんどん「inference.decisions」に叩き込む感じ。そして送信終わった各レコードのパーティション・オフセット管理も怠らない。ちゃんと最後のoffset+1をMap
それでね、Gradleには例の「org.apache.kafka:kafka-clients:3.7.0」これ記載が必須！プロデューサ側はちょっと細かく攻めるのがキモ。「bootstrap.servers」は"kafka:9092"。んで、「enable.idempotence」、忘れたら後悔する設定。あと「acks」はall設定確定。それから…ええと、「max.in.flight.requests.per.connection」が5固定、「retries」にはInteger.MAX_VALUEド直球突っ込む。でさ「transactional.id」を"inference-tx-1"みたいに毎Producerごと別々！初期化はもちろんproducer.initTransactions()無視禁止。
次コンシューマーね？そう、「kafka:9092」で繋いでOK。「group.id」は"inference-workers"に固定しちゃって平気。「enable.auto.commit」はfalse一択だわ。そして激推しなのが「isolation.level」を"read_committed"指定、ここほんと半端ダメ絶対ポイントだから！中途半端な未コミットデータとか怖すぎ。
で、本番ループ話いくよ？まずpoll(500ms)でバッチ吸い上げ→空なら即戻り。でもレコード入ってきたらproducer.beginTransaction()→爆速トランザクション開始。そのままループ回してkey（例えばpayment_id）もpayloadもゴソッて抜き出してrunModel(payload, "fraud-v42")実行。このrunModelさぁ、payload×model_version=判定結果、要するに完全な決定論型だと思う。
出てきたInferenceDecisionはserialize（直列化）してから新しいProducerRecord化→どんどん「inference.decisions」に叩き込む感じ。そして送信終わった各レコードのパーティション・オフセット管理も怠らない。ちゃんと最後のoffset+1をMap
的なのに全部詰め込んじゃう！
締め作業来たらsendOffsetsToTransaction(offsets, group.id)投げ、そのgroup.idは事前config指定あったアレ。そしてcommitTransaction()バン！終わり！！この流れ一本ですべて包み込み公開イメージね。でもエラー起こったら？producer.abortTransaction()押せばいい。自動的に重複送信まで防止する賢さ。
なんでこんな方法必要かというと…仮に書き込み済みだったけどオフセットコミット前とか超ハマる落とし穴でもworkerぶっ壊れても外から見れば綺麗サッパリ見えない仕掛けなんだよ！つまるところ常時整合性キープできる安心感、それが最高な部分！！ Read→Decide→Write→Commitモデルで推論ワークフローを強化しよう
- 再起動した時、入力データがもう一度処理されて同じアウトプットが返るパターンね。これ結構ありがちで、現場でもわりとみんなぶつかるやつだと思うな。
- ここでポイントなのは「Pythonゲートウェイのidempotent outbox」って話。もしモデルサーバーがFastAPIとかvLLM、TritonなどPython系の場合、とにかくネットワーク絡みの副作用を「ホットパス」から排除すべきなんだよ。つまり下流に流すなら、まず耐障害性重視でしっかりしたoutboxを書いておくべし、ということになる。
- 具体例として、このPythonサンプルコード見ると、「confluent_kafka」のProducer使ってて、「bootstrap.servers」は直書き（"kafka:9092"）、しかも「enable.idempotence」がTrue設定。そして「acks」が"all"になってる点も忘れずにチェックしたいな。
- decision_idを生成する関数も準備してあって、key・model_version・body_bytesをmmh3.hash_bytesで合成、その結果をhex変換する手順。handle_request(event_key, request_obj, model_version="fraud-v42")だと、1ステップ目はモデル推論処理でscore取得（インプット＋バージョンのみ依存＝純粋関数扱い）、そのあとdecisionデータの組み立てという流れ。
- 具体的にはevent_keyやmodel_version、それからscoreを全部json.dumps()してbody_bytes作成、その上でmmh3.hash_bytes→decision_id計算って段取りかな。それでKafka側はheaders=[("decision_id", did)]付きで「inference.decisions」トピックにproduceするようにしている。
- 注意点なんだけど、副作用処理（メール送信や支払い保留）はこの段階ではやらない。そのかわり、「decision_id」「action」「amount」など入れたoutboxイベントを別途まとめて「inference.outbox」に投げ込む方式にしてる。最後にp.flush()コールして一括書き込みする感じね。
- 一番大事なのはworker側。「decision_id」で冪等性チェックして、「既にこのID見たな」と思ったらスルー＝no-opする設計。このID単位で必ず一回だけ本処理される仕掛けなんだよね。
- あとTips的なものとして：モデルバージョンは1デプロイごと固定推奨＆payloadにも埋め込む。それから特徴量についても必ずスナップショット化やtime-travel方式を併用することで安全性アップ狙うのがいいと思う！まあ…time-travel featureについて語りたいけど今回は置いとくわ。
- ここでポイントなのは「Pythonゲートウェイのidempotent outbox」って話。もしモデルサーバーがFastAPIとかvLLM、TritonなどPython系の場合、とにかくネットワーク絡みの副作用を「ホットパス」から排除すべきなんだよ。つまり下流に流すなら、まず耐障害性重視でしっかりしたoutboxを書いておくべし、ということになる。
- 具体例として、このPythonサンプルコード見ると、「confluent_kafka」のProducer使ってて、「bootstrap.servers」は直書き（"kafka:9092"）、しかも「enable.idempotence」がTrue設定。そして「acks」が"all"になってる点も忘れずにチェックしたいな。
- decision_idを生成する関数も準備してあって、key・model_version・body_bytesをmmh3.hash_bytesで合成、その結果をhex変換する手順。handle_request(event_key, request_obj, model_version="fraud-v42")だと、1ステップ目はモデル推論処理でscore取得（インプット＋バージョンのみ依存＝純粋関数扱い）、そのあとdecisionデータの組み立てという流れ。
- 具体的にはevent_keyやmodel_version、それからscoreを全部json.dumps()してbody_bytes作成、その上でmmh3.hash_bytes→decision_id計算って段取りかな。それでKafka側はheaders=[("decision_id", did)]付きで「inference.decisions」トピックにproduceするようにしている。
- 注意点なんだけど、副作用処理（メール送信や支払い保留）はこの段階ではやらない。そのかわり、「decision_id」「action」「amount」など入れたoutboxイベントを別途まとめて「inference.outbox」に投げ込む方式にしてる。最後にp.flush()コールして一括書き込みする感じね。
- 一番大事なのはworker側。「decision_id」で冪等性チェックして、「既にこのID見たな」と思ったらスルー＝no-opする設計。このID単位で必ず一回だけ本処理される仕掛けなんだよね。
- あとTips的なものとして：モデルバージョンは1デプロイごと固定推奨＆payloadにも埋め込む。それから特徴量についても必ずスナップショット化やtime-travel方式を併用することで安全性アップ狙うのがいいと思う！まあ…time-travel featureについて語りたいけど今回は置いとくわ。
Java×Kafkaトランザクション活用で安全なリアルタイム推論を構築しよう
ポイント・イン・タイムでリード、うん、それやっとくとリプレイのズレ防げるっぽい。まあ当然っちゃ当然なんだけど、この部分わりと見落としがちだな。ちょっと重要。
それと、「exactly-once」つまり厳密に一度だけの処理を複数サービスで実現する時、アウトボックスパターン結構使えるよ。アプリは `inference.decisions` と `inference.outbox` って二箇所に書き込むんだよね…まあ仕組み自体はこれだけ。ただそれだけじゃ終わらなくて、コネクタとかワーカーが次に `inference.outbox` を読む。んで、例えば外部システムへのIDEMPOTENTアクセス──何回投げても同じになる奴──なら /holds/{decision_id} にPUT投げたり、とかそんな感じ。
あとKafka StreamsとかFlinkについてもちょっと。ストリーム上で特徴量JOINしたりローリング集約したいならステートストア＋トランザクション必要。KTableから読みつつ出力トピックへ書き込む、その両方を1コミット境界内に収める感じだな。そのおかげで単一決定保証ちゃんとキープできるっぽい。
…そういえば「リトライ」とか「スケール」、「ロングテール問題」とか気になった？まだそこ触れてないけど、その辺もまあ色々手当いるんだよなー、多分。
それと、「exactly-once」つまり厳密に一度だけの処理を複数サービスで実現する時、アウトボックスパターン結構使えるよ。アプリは `inference.decisions` と `inference.outbox` って二箇所に書き込むんだよね…まあ仕組み自体はこれだけ。ただそれだけじゃ終わらなくて、コネクタとかワーカーが次に `inference.outbox` を読む。んで、例えば外部システムへのIDEMPOTENTアクセス──何回投げても同じになる奴──なら /holds/{decision_id} にPUT投げたり、とかそんな感じ。
あとKafka StreamsとかFlinkについてもちょっと。ストリーム上で特徴量JOINしたりローリング集約したいならステートストア＋トランザクション必要。KTableから読みつつ出力トピックへ書き込む、その両方を1コミット境界内に収める感じだな。そのおかげで単一決定保証ちゃんとキープできるっぽい。
…そういえば「リトライ」とか「スケール」、「ロングテール問題」とか気になった？まだそこ触れてないけど、その辺もまあ色々手当いるんだよなー、多分。
Pythonとアウトボックスパターンで副作用の再発行リスクを抑えよう
えー、まあ分かる人多いと思うけどさ、「再試行って結局なくならないの？」みたいな話になるよね。うーん…実際、たぶんその疑問は普通に出てくるやつ。でも別に、それ自体は問題ないんだよな。だってトランザクション挟まってるし、再試行しても結局新しい何かが発生するわけじゃなくて、なんかただ同じ内容を何回も確認してるだけになるんだよね…。まあ、その辺分かりづらいけど。で、パーティショニングをビジネスキーでやるの超重要だと思う。これちゃんと横展開できるから。いや、本当にここ大事なんだよな、うん。
…でさ、モデルサーバーは絶対ステートレスの方がいいと思う。あー、これ割と勘違いされやすいというか、自分も最初間違えそうになったポイントなんだけど。でも実はホットな特徴量キャッシュはアリっちゃアリ。ただ、それ便利だからって勘違いしちゃダメでさ。本当に大事な事実が記録されてる場所じゃないんだわ、このキャッシュって。まじ気を付けてほしいな。
それと…レイテンシのトレードオフについてだけどさ、Exactly-once狙うとちょっとだけオーバーヘッド出る感じになるんだよね。例えば追加でラウンドトリップとか発生したり、トランザクションフェンス増えたりする…みたいな雰囲気？まあ正直、このへんは避けられないかなぁ。でも、安全性と引き換えって思えば、そこまで悪くない…かもしれないなぁ。
…でさ、モデルサーバーは絶対ステートレスの方がいいと思う。あー、これ割と勘違いされやすいというか、自分も最初間違えそうになったポイントなんだけど。でも実はホットな特徴量キャッシュはアリっちゃアリ。ただ、それ便利だからって勘違いしちゃダメでさ。本当に大事な事実が記録されてる場所じゃないんだわ、このキャッシュって。まじ気を付けてほしいな。
それと…レイテンシのトレードオフについてだけどさ、Exactly-once狙うとちょっとだけオーバーヘッド出る感じになるんだよね。例えば追加でラウンドトリップとか発生したり、トランザクションフェンス増えたりする…みたいな雰囲気？まあ正直、このへんは避けられないかなぁ。でも、安全性と引き換えって思えば、そこまで悪くない…かもしれないなぁ。
モデル・特徴量管理とサービス間連携パターンで重複レス設計を進めよう
うーん、まず言いたいのは…監査性があることでさ、不安なままリプレイしなくて済むみたいな？まあこれって、多分どのチームにとってもけっこうでかいメリットなんじゃないかな。損得考えると、あー、いやもう圧倒的に得かもしれないね。
えっと、出荷直前チェックみたいな感じで簡単に並べるね。
・キーはビジネスID一本で統一しといて、スキーマには絶対バージョン管理入れること。シンプルだけどマストだよ、忘れると色々面倒になるから。
・プロデューサー側は `enable.idempotence=true` を付ける。それで `acks=all` もオン。あと絶対に安定した `transactional.id` が必要…まあ、この辺ちゃんとやらないと意味ないから気を付けて。
コンシューマーについてなんだけど…えーと、`read_committed` を設定すること、それからオフセットコミットは必ず _トランザクション内_ じゃないとダメっぽいわ。
あと判断ロジックだけどね、その都度 `model_version` と決定論的な `decision_id` をちゃんとセットしておくのが無難というか、安全？
副作用関連もちょっと注意いるかな。すべて **outbox** ストリームを経由して、それから後ろの処理は冪等シンクだけ使うようにしてほしい感じ。
んー、再実行した場合どうなるかっていう話だよね。結果が全く同じアウトプットになるか、何にも起こらない――この２つどちらかしか発生しないよう設計されてれば、とりあえず大丈夫だと思うよ。
えっと、出荷直前チェックみたいな感じで簡単に並べるね。
・キーはビジネスID一本で統一しといて、スキーマには絶対バージョン管理入れること。シンプルだけどマストだよ、忘れると色々面倒になるから。
・プロデューサー側は `enable.idempotence=true` を付ける。それで `acks=all` もオン。あと絶対に安定した `transactional.id` が必要…まあ、この辺ちゃんとやらないと意味ないから気を付けて。
コンシューマーについてなんだけど…えーと、`read_committed` を設定すること、それからオフセットコミットは必ず _トランザクション内_ じゃないとダメっぽいわ。
あと判断ロジックだけどね、その都度 `model_version` と決定論的な `decision_id` をちゃんとセットしておくのが無難というか、安全？
副作用関連もちょっと注意いるかな。すべて **outbox** ストリームを経由して、それから後ろの処理は冪等シンクだけ使うようにしてほしい感じ。
んー、再実行した場合どうなるかっていう話だよね。結果が全く同じアウトプットになるか、何にも起こらない――この２つどちらかしか発生しないよう設計されてれば、とりあえず大丈夫だと思うよ。
Exactly-Once設計で再試行時も一貫したML結果と運用チェックリストを得よう
あー、Exactly-onceってよく魔法っぽいみたいに言われてるけど、正直そんなのじゃないんだよね。いや、実際はさ、まあ…単なる「規律」的なやつで。Kafkaとか、まあ一応色々プリミティブ渡してきてくれるんだけど、結局、それだけじゃ全然十分とは言えなくて…。大事なのって、自分たちのエンジニアとしての選択肢？この部分に尽きるかもしれないな。
たとえば安定なキー、あとはもう決定が全部先に決まってて絶対変わらんパターンとかね、副作用は後からでも全部追えるように記録残すとか。そこが重要そう、うーん…。
それから、とりあえず何もかも最初からフルスピードでやる必要ない気がするわ。最初はホント簡単な構成だけ動かせばいいんじゃないかな。一個のサービスにトランザクションまとめちゃってさ、outboxパターン当て込む。それでまずは1日分データとか思いきってリプレイして試してみる、と。
それで出力内容変わらないなら「ああ〜、たぶん合ってるっぽい」みたいになるかな…多分、その進め方悪くはないんじゃないかなぁ、と私は思うわ。
たとえば安定なキー、あとはもう決定が全部先に決まってて絶対変わらんパターンとかね、副作用は後からでも全部追えるように記録残すとか。そこが重要そう、うーん…。
それから、とりあえず何もかも最初からフルスピードでやる必要ない気がするわ。最初はホント簡単な構成だけ動かせばいいんじゃないかな。一個のサービスにトランザクションまとめちゃってさ、outboxパターン当て込む。それでまずは1日分データとか思いきってリプレイして試してみる、と。
それで出力内容変わらないなら「ああ〜、たぶん合ってるっぽい」みたいになるかな…多分、その進め方悪くはないんじゃないかなぁ、と私は思うわ。
