kafka-development

Kafka 開発

このスキルでは、Apache Kafka によるイベントストリーミングと分散メッセージングシステムのベストプラクティスを紹介します。Kafka ベースのアプリケーションを構築する際に、これらのガイドラインを適用してください。

基本原則

• Kafka は、高スループットでフォールトトレラントなメッセージングを実現する分散イベントストリーミングプラットフォームです
• 従来の pub/sub とは異なり、Kafka はプルモデルを使用します。コンシューマーがパーティションからメッセージを取得します
• 必要に応じて、スケーラビリティ、耐久性、厳密に1回のセマンティクスを考慮して設計してください
• 実装に未対応事項、プレースホルダー、未完成の部分を残さないでください

ワークフロー：Kafka プロデューサー・コンシューマーパイプラインの設定

1. トピックを定義する — 説明的な名前を選び、想定されるコンシューマーの並列度に基づいてパーティション数を設定し、保持期間とレプリケーション係数を構成します。
2. メッセージスキーマを設計する — Schema Registry に Avro または JSON スキーマを登録し、最初から後方互換性を確保します。
3. プロデューサーを実装する — acks=all を設定し、べき等性を有効にし、均等に分散されるパーティションキーを選択して、再試行ロジックを含むエラー処理を追加します。
4. コンシューマーを実装する — enable.auto.commit=false を設定し、適切な auto.offset.reset ポリシーを選び、メッセージをべき等に処理して、処理の成功後にのみオフセットをコミットします。
5. 可観測性を追加する — プロデューサーの送信レート、コンシューマーのラグ、ブローカーのレプリカ不足パーティションを計測し、メッセージヘッダーでトレースコンテキストを伝播させます。
6. エンドツーエンドでテストする — Testcontainers または埋め込み Kafka ブローカーを使用し、障害やリバランスのシナリオも含めて、生成から消費、コミットまでの全サイクルを検証します。
7. デプロイして監視する — ラグアラート、継続的な失敗に対するデッドレタートピックへのルーティング、主要なブローカーおよびクライアントのメトリクス用ダッシュボードを設定して、段階的に展開します。

アーキテクチャの概要

中核コンポーネント

• トピック：メッセージを整理するためのカテゴリ／フィード
• パーティション：並列処理を可能にする、トピック内の順序付けられた不変のシーケンス
• プロデューサー：トピックにメッセージを発行するクライアント
• コンシューマー：トピックからメッセージを読み取るクライアント
• コンシューマーグループ：複数のコンシューマー間での消費を調整します
• ブローカー：データを保存し、クライアントにサービスを提供する Kafka サーバー

重要な概念

• メッセージはパーティションに順番に追加されます
• 各メッセージにはオフセットがあり、パーティション内で一意の連番 ID です
• コンシューマーは独自のカーソル（オフセット）を管理し、ストリームを繰り返し読み取れます
• スケーラビリティを確保するため、パーティションはブローカー間で分散されます

トピック設計

パーティショニング戦略

• 関連するイベントを同じパーティションに配置するには、パーティションキーを使用します
• 同じキーを持つメッセージは、常に同じパーティションに送られます
• これにより、関連するイベントの順序が保証されます
• キーは慎重に選択してください。分布が不均等だと、ホットパーティションが発生します

パーティション数

• パーティション数が多いほど並列度は高まりますが、オーバーヘッドも増加します
• 想定スループット、コンシューマー数、ブローカーのリソースを考慮します
• 同時に実行する想定のコンシューマー数を初期値にします
• パーティション数は増やせますが、減らすことはできません

トピックの設定

• retention.ms：メッセージを保持する期間（デフォルトは7日）
• retention.bytes：パーティションあたりの最大サイズ
• cleanup.policy：delete（古いものを削除）または compact（キーごとに最新のものを保持）
• min.insync.replicas：確認応答が必要なレプリカの最小数

プロデューサーのベストプラクティス

信頼性の設定

acks=all               # Wait for all replicas to acknowledge
retries=MAX_INT        # Retry on transient failures
enable.idempotence=true # Prevent duplicate messages on retry

パフォーマンスの調整

• batch.size：送信前にメッセージを蓄積するサイズ（デフォルトは16KB）
• linger.ms：バッチ処理の待機時間（0 = 即時送信）
• buffer.memory：未送信メッセージをバッファリングするためのメモリ総量
• compression.type：帯域幅を節約するための gzip、snappy、lz4、または zstd

エラー処理

• 指数バックオフを用いた再試行ロジックを実装します
• 再試行可能な例外と再試行不可能な例外を、それぞれ異なる方法で処理します
• 送信の失敗をログに記録し、アラートを通知します
• 繰り返し失敗するメッセージには、デッドレタートピックを検討します

例：べき等性とエラー処理を備えた Java プロデューサー

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
    ProducerRecord<String, String> record =
        new ProducerRecord<>("orders", "order-123", "{\"item\":\"widget\",\"qty\":5}");

    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            log.error("Send failed for key=order-123", exception);
            // Route to dead-letter topic or alert
        } else {
            log.info("Delivered to {}-{} offset {}",
                metadata.topic(), metadata.partition(), metadata.offset());
        }
    });
}

パーティショナー

• デフォルト：キーのハッシュ値によってパーティションが決まります（キーが null の場合はラウンドロビン）
• 特定のルーティング要件に応じたカスタムパーティショナー
• ホットパーティションを避けるため、均等に分散されるようにします

コンシューマーのベストプラクティス

オフセット管理

• コンシューマーは、オフセットを使って処理済みメッセージを追跡します
• auto.offset.reset：earliest（最初から開始）または latest（新しいメッセージのみ）
• 正常に処理した後にオフセットをコミットし、処理前にはコミットしないでください
• exactly-once セマンティクスを実現するには、enable.auto.commit=false を使用します

コンシューマーグループ

• グループ内のコンシューマーでパーティションを分担します（各パーティションは 1 つのコンシューマーが担当します）
• パーティション数よりコンシューマー数が多い場合、一部のコンシューマーはアイドル状態になります
• コンシューマーが参加または離脱すると、グループのリバランスが発生します
• リバランスを減らすには、静的メンバーシップに group.instance.id を使用します

処理パターン

• パーティション内ではメッセージを順番に処理します
• 必要に応じて、パーティション間で順序が入れ替わったメッセージを処理します
• at-least-once 配信に対応するため、冪等な処理を実装します
• exactly-once を実現するには、トランザクション処理を検討します

例：オフセットを手動でコミットする Java コンシューマー

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Collections.singletonList("orders"));

    while (running) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
        for (ConsumerRecord<String, String> record : records) {
            try {
                processOrder(record.key(), record.value());
            } catch (Exception e) {
                log.error("Failed to process offset={} key={}", record.offset(), record.key(), e);
                publishToDeadLetterTopic(record, e);
            }
        }
        consumer.commitSync(); // Commit only after successful processing
    }
}

タイムアウトと障害

• 処理タイムアウトを実装して、処理の遅いイベントを切り離します
• タイムアウトが発生した場合、イベントを別にして次のメッセージに進みます
• すべてのイベントを 1 件ずつ処理することよりも、システム全体のパフォーマンスを維持します
• すべての再試行に失敗したメッセージには、デッドレターキューを使用します

エラー処理と再試行

再試行戦略

• 1 回の処理試行につき、複数回のランタイム再試行を許可します
• 例：redrive ごとにランタイム再試行を 3 回、最大 5 回 redrive すると、再試行は合計 15 回です
• ランタイム再試行で通常、障害の 99% に対応できます
• 再試行を使い切った後は、デッドレターキューにルーティングします

デッドレタートピック

• 処理できないメッセージ用に、専用の DLT を作成します
• 元のトピック、パーティション、オフセット、エラーの詳細を含めます
• システム全体の問題を示すパターンがないか、DLT を監視します
• DLT からの手動または自動の再試行を実装します

スキーマ管理

スキーマレジストリ

• スキーマ管理には Confluent Schema Registry を使用します
• プロデューサーはシリアライズ時に、登録済みのスキーマに対してデータを検証します
• スキーマが一致しない場合は例外が発生し、不正なデータが防止されます
• プロデューサーとコンシューマーの共通の参照先となります

スキーマの進化

• 前方互換性と後方互換性を考慮してスキーマを設計します
• 後方互換性を保つには、デフォルト値を持つオプションフィールドを追加します
• フィールドの削除や名前の変更は避けます
• スキーマのバージョン管理と移行戦略を使用します

Kafka Streams

状態管理

• ログコンパクションを実装して、各キーの最新バージョンを維持します
• 状態ストアから古いデータを定期的に削除します
• 状態ストアのサイズとアクセスパターンを監視します
• システムの規模に適したストレージバックエンドを使用します

ウィンドウ処理

• 順序が入れ替わったイベントや、偏ったタイムスタンプに対応します
• 適切な時刻抽出とウォーターマークの手法を使用します
• 遅れて到着するデータに対する猶予期間を設定します
• ユースケースに応じてウィンドウの種類（タンブリング、ホッピング、スライディング、セッション）を選択します

セキュリティ

認証

• クライアント認証には SASL/SSL を使用します
• SASL メカニズムとして、PLAIN、SCRAM、OAUTHBEARER、GSSAPI に対応します
• 通信中の暗号化には SSL を有効にします
• 認証情報を定期的にローテーションします

認可

• きめ細かなアクセス制御には Kafka ACL を使用します
• プリンシパルごとに必要最小限の権限を付与します
• トピックごとに読み取り権限と書き込み権限を分けます
• アクセスパターンを定期的に監査します

モニタリングと可観測性

主要なメトリクス

• プロデューサー：record-send-rate、record-error-rate、batch-size-avg
• コンシューマー：records-consumed-rate、records-lag、commit-latency
• ブローカー：under-replicated-partitions、request-latency、disk-usage

ラグのモニタリング

• コンシューマーラグ = 最後に生成されたオフセット - 最後にコミットされたオフセット
• ラグが大きい場合、コンシューマーが処理に追いついていないことを示します
• ラグが増加する傾向に対してアラートを設定します
• コンシューマーをスケールするか、処理を最適化します

分散トレーシング

• メッセージヘッダーでトレースコンテキストを伝播させます
• エンドツーエンドのトレーシングにはOpenTelemetryを使用します
• プロデューサーとコンシューマーのスパンを関連付けます
• パイプライン内のメッセージの経路を追跡します

テスト

単体テスト

• 分離してテストするためにKafkaクライアントをモックします
• シリアライズ／デシリアライズのロジックをテストします
• パーティショニングのロジックを検証します
• エラー処理の経路をテストします

統合テスト

• 組み込みKafkaまたはTestcontainersを使用します
• プロデューサーからコンシューマーまでの一連のフローをテストします
• 使用している場合は、厳密に1回のセマンティクスを検証します
• リバランスのシナリオをテストします

パフォーマンステスト

• 本番環境に近いメッセージレートで負荷テストを実施します
• コンシューマーのスループットとラグの挙動をテストします
• 負荷時のブローカーのリソース使用量を検証します
• 障害と復旧のシナリオをテストします

一般的なパターン

イベントソーシング

• すべての状態変更を不変のイベントとして保存します
• イベントを再生して状態を再構築します
• スナップショットにはログコンパクションを使用します
• タイムトラベルデバッグを可能にします

CQRS（コマンドクエリ責務分離）

• 書き込み（コマンド）モデルと読み取り（クエリ）モデルを分離します
• Kafkaをイベントストアとして使用します
• イベントから読み取りに最適化されたプロジェクションを構築します
• 適切に結果整合性を扱います

サーガパターン

• サービス間の分散トランザクションを調整します
• 各サービスは次のステップに向けてイベントを発行します
• ロールバック用に補償トランザクションを実装します
• 相関IDを使用してサーガのインスタンスを追跡します

変更データキャプチャ（CDC）

• データベースの変更をKafkaイベントとしてキャプチャします
• DebeziumなどのCDCツールを使用します
• リアルタイムのデータ同期を可能にします
• イベント駆動型の統合を構築します