1. 実際の課題：トランザクション集中によるインフラ崩壊
オンラインチケット販売サイトを運営しているとします。普段は1日にパラパラと数十件程度の決済しか発生せず、すべてが極めてスムーズに稼働しています。
しかし、限定VIPチケットの発売開始日時である午前10時、悲劇が起こります。およそ2万人のユーザーが、まったく同じ1秒間に一斉に「購入」ボタンをクリックしたのです。
その影響は瞬く間に広がりました。決済サーバーは過負荷に陥り、CPU使用率は100%に急上昇。在庫サービス側の更新が追いつかず、同一チケットの二重販売が発生してしまいます。さらに深刻なことに、各マイクロサービスがHTTPによる同期呼び出し（AがBを呼び出し、BがCの応答を待つ）を行っていたため、わずか1つのサービスが3秒遅延しただけで、処理チェーン全体が一瞬にして完全に目詰まりを起こしました。
画面のローディングが止まらないのを見たユーザーは、ブラウザのF5キー（再読み込み）を連打します。これによりアクセス数は一気に5倍に跳ね上がり、メインデータベースのコネクションプールは完全に枯渇、最終的にシステム全体が応答不能に陥ってしまいました。
2. 原因の分析：密結合の罠とZooKeeperの運用負荷
この障害は、アーキテクチャにおける2つの根本的な弱点に起因しています。
- 過度な密結合（Tight Coupling）： 注文を確定させるために、在庫サービスやメール通知サービスが常時オンラインでなければならない設計になっていました。メールサービスが詰まるとチケット購入処理そのものがフリーズします。このようなアーキテクチャでは、局所的な障害に耐えるフォールトトレランスが完全に失われてしまいます。
- ストレージツールの誤選定： リレーショナルデータベース（RDBMS）は、最終的なステータス（状態）を保持するために作られています。1秒間に2万〜3万件も押し寄せるイベントを直接受け止めるのには向いていません。
これまで、この種のデータストリームを吸収する標準的な解決策はApache Kafkaの導入でした。しかし、従来の構成にはもう一つの重荷が付きまとっていました。それがApache ZooKeeperです。
夜間オンコールを経験したエンジニアなら、Kafkaと並行してZooKeeperクラスターを保守・運用する苦労がよく分かるはずです。ZooKeeperは独自のJVMで稼働し、各ノードで最低でも2〜4GBのメモリを余計に消費する上、個別のバックアップ体制も必要です。さらに、ネットワーク分断（ネットワークパーティション）が発生した際、ZooKeeperとKafkaブローカー間のメタデータ同期にズレが生じ、クラスターの復旧に膨大な時間を費やすこともしばしばありました。
3. 代表的なアプローチの比較
データ処理のボトルネックを解消するため、エンジニアは通常以下の3つのアプローチを検討します。
- データベースのスケールアップ（Vertical Scaling）： メモリを32GBから128GBに増設し、エンタープライズ向けNVMeストレージを採用する手法です。目先の症状は一時的に緩和されますが、インフラコストは3倍に跳ね上がります。また、ボトルネックが次のチケット販売イベントへ先送りされるに過ぎません。
- 従来のメッセージキューの利用（RabbitMQ、Redis Pub/Sub）： サービス間の疎結合化には非常に有効です。しかし、イベントの再処理（イベントリプレイ）のためにログを30日間保持したい場合や、スループットが毎秒10万メッセージを超えるような状況では、RabbitMQなどのメモリ消費が限界に達し始めます。
- KRaftモードでのKafka運用： Kafkaが誇る毎秒数百万件規模の高スループット処理能力はそのままに、仲介役となっていたZooKeeper層を完全に排除します。
4. 解決策：CentOS Stream 9でKRaftモードのApache Kafkaを構築する
バージョン3.3以降、KRaft（Kafka Raft Metadata Mode）は正式に本番環境（プロダクション）対応となりました。Kafka自身がコントローラーを選出し、Raft合意アルゴリズムを用いて内部でメタデータを直接管理します。クラスターの起動時間は数分からわずか数秒へと短縮され、数百万におよぶパーティションの管理も軽快に行えます。
ステップ1：環境準備とJava 17のインストール
KafkaはJVM上で動作します。CentOS Stream 9では、長期サポート（LTS）版であるOpenJDK 17が安定して利用可能です。
sudo dnf update -y
sudo dnf install -y java-17-openjdk-headless wget tar
java -versionセキュリティ強化のため、対話型シェルへのログイン権限を持たない専用のシステムユーザーを作成します：
sudo useradd -r -d /opt/kafka -s /sbin/nologin kafkaステップ2：Apache Kafkaのダウンロードと展開
Kafkaのバイナリパッケージをダウンロードします（Scala 2.13バンドル済みのバージョン3.7.0を推奨）：
cd /tmp
wget https://archive.apache.org/dist/kafka/3.7.0/kafka_2.13-3.7.0.tgz
sudo tar -xzf kafka_2.13-3.7.0.tgz -C /opt
sudo mv /opt/kafka_2.13-3.7.0 /opt/kafka
sudo chown -R kafka:kafka /opt/kafkaステップ3：Kafka KRaftの設定（単一ノード構成）
KRaftモードにおけるメイン設定ファイルは、/opt/kafka/config/kraft/server.propertiesに配置されています。このファイルを編集します：
sudo vi /opt/kafka/config/kraft/server.properties以下のように基本パラメータを設定します（192.168.1.50はサーバーのLAN/パブリックIPアドレスに置き換えてください）：
# ブローカーとメタデータコントローラーを兼ねるノードとして指定
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@localhost:9093
# リスナーの待ち受けポート設定
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://localhost:9093
advertised.listeners=PLAINTEXT://192.168.1.50:9092
controller.listener.names=CONTROLLER
# メッセージログの保存先ディレクトリおよび保持期間（7日間）
log.dirs=/var/lib/kafka/data
num.partitions=3
log.retention.hours=168データログ保存用ディレクトリを作成し、kafkaユーザーに所有権を付与します：
sudo mkdir -p /var/lib/kafka/data
sudo chown -R kafka:kafka /var/lib/kafkaステップ4：Storage Cluster IDの初期化
初回起動の前に、KRaftでは一意なCluster IDを用いてストレージディレクトリをフォーマットする必要があります：
# クラスター識別用UUIDの生成
KAFKA_CLUSTER_ID=$(sudo -u kafka /opt/kafka/bin/kafka-storage.sh random-cluster-id)
echo "クラスターID: $KAFKA_CLUSTER_ID"
# クラスターIDを使用してログディレクトリをフォーマット
sudo -u kafka /opt/kafka/bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c /opt/kafka/config/kraft/server.propertiesステップ5：プロセス管理用Systemd Serviceの作成
サーバー再起動時にもCentOS Stream 9とともにKafkaが自動起動するよう、Serviceユニットファイルを作成します：
sudo tee /etc/systemd/system/kafka.service << 'EOF'
[Unit]
Description=Apache Kafka Server (KRaft mode)
Documentation=https://kafka.apache.org/documentation/
After=network.target
[Service]
Type=simple
User=kafka
Group=kafka
Environment="JAVA_HOME=/usr/lib/jvm/jre-17-openjdk"
Environment="KAFKA_LOG_DIR=/opt/kafka/logs"
Environment="KAFKA_HEAP_OPTS=-Xmx1G -Xms1G"
ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/kraft/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=on-failure
LimitNOFILE=65536
[Install]
WantedBy=multi-user.target
EOFデーモンをリロードしてサービスを起動・自動有効化します：
sudo systemctl daemon-reload
sudo systemctl enable --now kafka
sudo systemctl status kafkaステップ6：Firewalldでポート開放
ファイアウォールが有効な場合は、クライアント接続用にポート9092を開放します：
sudo firewall-cmd --permanent --add-port=9092/tcp
sudo firewall-cmd --reloadステップ7：データ送受信パイプラインの検証
標準同梱されているCLIツールを使って、メッセージの送受信パイプラインが正常に疎通しているかテストします。
3つのパーティションを持つorders-streamというテスト用トピックを作成します：
/opt/kafka/bin/kafka-topics.sh --create --topic orders-stream --partitions 3 --replication-factor 1 --bootstrap-server localhost:9092Console Producerを起動し、2件のJSONテストデータを送信してみます：
/opt/kafka/bin/kafka-console-producer.sh --topic orders-stream --bootstrap-server localhost:9092
>{"order_id": 1001, "amount": 250000, "status": "PENDING"}
>{"order_id": 1002, "amount": 500000, "status": "PAID"}別のターミナルタブを開き、トピックの先頭からデータを読み取るConsole Consumerを実行します：
/opt/kafka/bin/kafka-console-consumer.sh --topic orders-stream --from-beginning --bootstrap-server localhost:9092Consumer側の画面に、先ほど送信した2件のJSONメッセージが即座に出力されます。これでデータがディスクへ永続化されたことが確認できました。たとえConsumerに障害が発生して数時間停止したとしても、すべてのメッセージはKafkaブローカー上に安全に保持され、いつでも再処理が可能です。
