# 機能設計書 31-高可用性

## 概要

本ドキュメントは、Apache Flinkの高可用性（High Availability: HA）機能について記載する。この機能は、分散ストリーム処理システムにおいてJobManagerやResourceManagerなどの重要コンポーネントの障害に対する耐障害性を提供する。

### 本機能の処理概要

高可用性機能は、Flinkクラスタにおけるマスターコンポーネント（JobManager、ResourceManager、Dispatcher）の障害発生時に、自動的にフェイルオーバーを行い、ジョブの継続実行を保証する仕組みを提供する。

**業務上の目的・背景**：分散ストリーム処理システムでは、24時間365日の連続稼働が求められる。マスターコンポーネントの障害によるジョブの停止は、ビジネス上の損失につながるため、自動復旧機能が必須である。高可用性機能により、単一障害点を排除し、システム全体の信頼性を向上させる。

**機能の利用シーン**：本番環境での長期間運用、金融システムやリアルタイム分析基盤など、ダウンタイムを許容できないミッションクリティカルなシステムで利用される。ZooKeeperまたはKubernetesと連携して、リーダー選出やメタデータの永続化を行う。

**主要な処理内容**：
1. リーダー選出（Leader Election）：複数のJobManager候補から1つをアクティブリーダーとして選出
2. リーダー検出（Leader Retrieval）：クライアントやワーカーがアクティブリーダーを発見
3. チェックポイントメタデータの永続化：チェックポイント情報をHA対応ストレージに保存
4. 実行計画（ExecutionPlan）の永続化：ジョブの実行計画を分散ストレージに保存
5. ジョブ結果の管理：完了したジョブの結果をJobResultStoreに保存

**関連システム・外部連携**：
- ZooKeeper：リーダー選出、メタデータ管理、分散ロック
- Kubernetes：ConfigMapを利用したリーダー選出、メタデータ管理
- 分散ファイルシステム（HDFS、S3など）：BLOBストア、ジョブ結果ストア

**権限による制御**：特定のロールによる制御はなく、クラスタ設定によって高可用性モードが決定される。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能は画面とは直接関連しない（バックエンド処理） |

## 機能種別

データ連携 / 分散協調 / 状態管理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| high-availability | String | No | HAモード（NONE/ZOOKEEPER/KUBERNETES/FACTORY_CLASS） | 列挙値チェック |
| high-availability.storageDir | String | Yes（HA有効時） | HAメタデータの保存先パス | 有効なファイルシステムパス |
| high-availability.cluster-id | String | Yes（HA有効時） | クラスタ識別子 | 空文字でないこと |
| high-availability.zookeeper.quorum | String | Yes（ZK使用時） | ZooKeeperクォーラムアドレス | 有効なホスト:ポート形式 |
| high-availability.zookeeper.path.root | String | No | ZooKeeper上のルートパス | 有効なZKパス |

### 入力データソース

- Flink設定ファイル（flink-conf.yaml）
- 環境変数
- コマンドライン引数

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| リーダー情報 | LeaderInformation | 選出されたリーダーのアドレスとセッションID |
| ジョブ結果 | JobResultEntry | 完了ジョブのID、結果ステータス、例外情報 |
| チェックポイントメタデータ | CheckpointMetadata | チェックポイントID、状態ハンドル |

### 出力先

- ZooKeeperノード（リーダー情報、チェックポイントカウンタ）
- 分散ファイルシステム（ジョブグラフ、チェックポイント、BLOB）
- Kubernetes ConfigMap（リーダー情報、メタデータ）

## 処理フロー

### 処理シーケンス

```
1. HAサービス初期化
   └─ 設定からHAモードを読み取り、適切なHAサービス実装をインスタンス化

2. リーダー選出サービス開始
   └─ ZooKeeper/Kubernetesにリーダー選出リクエストを登録

3. リーダーに選出された場合
   └─ リーダー情報をZK/K8sに書き込み、サービス開始

4. リーダー検出サービス
   └─ クライアント/ワーカーがリーダー情報を取得

5. チェックポイント・ジョブグラフ永続化
   └─ 各種メタデータを分散ストレージに保存

6. 障害検知・フェイルオーバー
   └─ リーダー障害時、新リーダーを選出しサービス復旧
```

### フローチャート

```mermaid
flowchart TD
    A[クラスタ起動] --> B{HAモード判定}
    B -->|NONE| C[StandaloneHaServices]
    B -->|ZOOKEEPER| D[ZooKeeperLeaderElectionHaServices]
    B -->|KUBERNETES| E[KubernetesHaServices]
    B -->|FACTORY_CLASS| F[カスタムHAサービス]

    D --> G[リーダー選出開始]
    G --> H{リーダーに選出?}
    H -->|Yes| I[サービス開始]
    H -->|No| J[スタンバイ待機]

    I --> K[メタデータ永続化]
    J --> L[リーダー監視]
    L --> M{リーダー障害?}
    M -->|Yes| G
    M -->|No| L
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-31-01 | HAモード必須 | 本番環境では必ずZOOKEEPERまたはKUBERNETESモードを使用 | 本番デプロイ時 |
| BR-31-02 | ストレージパス必須 | HA有効時はstorageDir設定が必須 | HA有効時 |
| BR-31-03 | クラスタID一意性 | 同一ZK/K8sを共有するクラスタは異なるcluster-idを使用 | 複数クラスタ運用時 |

### 計算ロジック

該当なし

## データベース操作仕様

### 操作別データベース影響一覧

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | RDBMSは使用しない。分散ストレージを使用 |

### テーブル別操作詳細

該当なし（ZooKeeper/Kubernetes/分散ファイルシステムを使用）

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| HA-001 | 設定エラー | storageDir未設定 | 有効なstorageDir設定を追加 |
| HA-002 | 接続エラー | ZooKeeper接続失敗 | ZKクォーラム設定確認、ネットワーク確認 |
| HA-003 | リーダー選出失敗 | リーダー選出タイムアウト | ZK/K8s状態確認、ログ確認 |
| HA-004 | ストレージエラー | 分散ストレージ書き込み失敗 | ストレージ権限・容量確認 |

### リトライ仕様

- ZooKeeper接続：Curatorフレームワークによる自動リトライ（指数バックオフ）
- ストレージ書き込み：設定可能なリトライ回数

## トランザクション仕様

分散協調処理のため、ZooKeeperのZxidに基づく楽観的同時実行制御を使用。メタデータ更新は原子的に実行される。

## パフォーマンス要件

- リーダー選出：ZooKeeperセッションタイムアウト内（デフォルト60秒）
- フェイルオーバー：数秒〜数十秒（設定・負荷依存）

## セキュリティ考慮事項

- ZooKeeper認証（SASL/Kerberos対応）
- TLS暗号化通信
- ACLによるZKノードアクセス制御

## 備考

- ZooKeeperモードではApache Curator 5.xを使用
- KubernetesモードではConfigMapベースのリーダー選出を使用

---

## コードリーディングガイド

本機能を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: データ構造を理解する

HAサービスで使用される主要なインターフェースと設定クラスを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | HighAvailabilityServices.java | `flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/` | HAサービスのメインインターフェース、各種サービス取得メソッド |
| 1-2 | JobResultStore.java | `flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/` | ジョブ結果ストアのインターフェース定義 |
| 1-3 | JobResultEntry.java | `flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/` | ジョブ結果エントリのデータ構造 |

**読解のコツ**: HighAvailabilityServicesインターフェースが提供する各種サービス（LeaderElection、LeaderRetrieval、CheckpointRecovery等）の役割を把握する。

#### Step 2: エントリーポイントを理解する

HAサービスのファクトリとユーティリティクラスを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | HighAvailabilityServicesUtils.java | `flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/` | HAサービス生成のファクトリメソッド群 |

**主要処理フロー**:
1. **59-84行目**: createAvailableOrEmbeddedServices()でHAモードに応じたサービスを生成
2. **86-96行目**: createZooKeeperHaServices()でZooKeeper HAサービスを生成
3. **148-171行目**: createClientHAService()でクライアント用HAサービスを生成

#### Step 3: 抽象基底クラスを理解する

共通処理を提供する抽象クラスを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | AbstractHaServices.java | `flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/` | 共通処理とテンプレートメソッド |

**主要処理フロー**:
- **72-85行目**: コンストラクタでの依存関係注入
- **88-131行目**: リーダー選出/検出サービスの生成
- **154-181行目**: close()メソッドでのリソース解放
- **183-210行目**: cleanupAllData()でのデータクリーンアップ

#### Step 4: ZooKeeper実装を理解する

ZooKeeperベースのHA実装を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | ZooKeeperLeaderElectionHaServices.java | `flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/` | ZooKeeperを使用したHA実装 |

**主要処理フロー**:
- **44-61行目**: ZKノード構造のコメント（重要）
- **66-82行目**: コンストラクタでのZK接続確立
- **85-91行目**: チェックポイントリカバリファクトリ生成
- **175-181行目**: リーダー検出サービス生成

#### Step 5: ジョブ結果ストアを理解する

ジョブ結果の永続化処理を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | FileSystemJobResultStore.java | `flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/` | ファイルシステムベースのジョブ結果ストア |

**主要処理フロー**:
- **63-64行目**: ファイル拡張子定義（.json、_DIRTY.json）
- **156-167行目**: createDirtyResultInternal()でダーティ結果作成
- **170-185行目**: markResultAsCleanInternal()でクリーンマーク

### プログラム呼び出し階層図

```
HighAvailabilityServicesUtils.createAvailableOrEmbeddedServices()
    │
    ├─ HighAvailabilityMode.fromConfig()
    │      └─ 設定からHAモード判定
    │
    ├─ createZooKeeperHaServices()
    │      ├─ ZooKeeperUtils.startCuratorFramework()
    │      └─ new ZooKeeperLeaderElectionHaServices()
    │             ├─ ZooKeeperLeaderElectionDriverFactory
    │             └─ FileSystemJobResultStore.fromConfiguration()
    │
    ├─ createCustomHAServices() [Kubernetes]
    │      └─ KubernetesHaServicesFactory.createHAServices()
    │
    └─ new EmbeddedHaServices() [NONE mode]
           └─ インメモリHA（テスト用）
```

### データフロー図

```
[入力]                    [処理]                         [出力]

flink-conf.yaml ───────▶ HighAvailabilityServicesUtils ───▶ HAサービスインスタンス
                               │
                               ▼
                         AbstractHaServices
                               │
        ┌──────────────────────┼──────────────────────┐
        ▼                      ▼                      ▼
 LeaderElection          JobResultStore        CheckpointRecovery
        │                      │                      │
        ▼                      ▼                      ▼
 ZooKeeper/K8s          分散ファイルシステム      ZooKeeper
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| HighAvailabilityServices.java | `flink-runtime/.../highavailability/` | インターフェース | HAサービスのメインAPI |
| AbstractHaServices.java | `flink-runtime/.../highavailability/` | 抽象クラス | 共通処理の実装 |
| HighAvailabilityServicesUtils.java | `flink-runtime/.../highavailability/` | ユーティリティ | HAサービスファクトリ |
| ZooKeeperLeaderElectionHaServices.java | `flink-runtime/.../highavailability/zookeeper/` | 実装クラス | ZooKeeper HA実装 |
| CuratorFrameworkWithUnhandledErrorListener.java | `flink-runtime/.../highavailability/zookeeper/` | ラッパー | Curatorエラーハンドリング |
| JobResultStore.java | `flink-runtime/.../highavailability/` | インターフェース | ジョブ結果ストアAPI |
| FileSystemJobResultStore.java | `flink-runtime/.../highavailability/` | 実装クラス | FS上のジョブ結果ストア |
| StandaloneHaServices.java | `flink-runtime/.../highavailability/nonha/standalone/` | 実装クラス | 非HAモード実装 |
| EmbeddedHaServices.java | `flink-runtime/.../highavailability/nonha/embedded/` | 実装クラス | 組み込みHA（テスト用） |
| HighAvailabilityOptions.java | `flink-core/` | 設定 | HA関連設定オプション |
