# 機能設計書 24-タスク実行

## 概要

本ドキュメントは、Apache Flinkのタスク実行機能（flink-runtime モジュール内TaskExecutor/TaskManager）の設計仕様を記述する。タスク実行機能は、Flinkジョブの個々のタスクを実際に実行するワーカーコンポーネントが提供する。

### 本機能の処理概要

TaskExecutor（TaskManager）は、Flinkクラスタのワーカーノードとして動作し、JobMasterから受け取ったタスクを実行する。メモリ管理、ネットワーク通信、ステート管理のローカル処理を担当する。

**業務上の目的・背景**：分散ストリーム/バッチ処理において、実際の演算処理（map、filter、aggregate等）を実行するワーカーコンポーネントを提供する。クラスタ全体のスケーラビリティは、TaskExecutorの数によって決定される。

**機能の利用シーン**：
- Flinkジョブのタスク実行
- データのシリアライゼーション/デシリアライゼーション
- ネットワーク経由のデータシャッフル
- ローカルステートの管理とチェックポイント

**主要な処理内容**：
1. ResourceManagerへの登録とハートビート
2. JobMasterへのスロット提供
3. タスクのデプロイと実行
4. タスク間のデータ交換（シャッフル）
5. チェックポイントへの参加
6. タスクステータスのレポート

**関連システム・外部連携**：
- ResourceManager（リソース登録）
- JobMaster（タスク受信）
- 他のTaskExecutor（データ交換）
- チェックポイントストレージ

**権限による制御**：InstanceIDによる識別、フェンシングトークンによる不正アクセス防止。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 31 | TaskManager一覧 | 主機能 | TaskManager一覧の取得 |
| 32 | TaskManager詳細 | 補助機能 | TaskManager詳細情報の取得 |
| 33 | TaskManagerメトリクス | 補助機能 | JVMメモリ、GC情報表示 |
| 36 | TaskManagerスレッドダンプ | 補助機能 | スレッド情報の取得 |

## 機能種別

計算処理 / データ処理 / リソース管理

## 入力仕様

### 入力パラメータ

#### タスクデプロイ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| taskDeploymentDescriptor | TaskDeploymentDescriptor | Yes | タスク情報 | 有効なタスク定義 |
| jobId | JobID | Yes | ジョブID | UUID形式 |
| allocationId | AllocationID | Yes | スロット割当ID | UUID形式 |

#### チェックポイント

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| executionAttemptID | ExecutionAttemptID | Yes | 実行試行ID | 有効なID |
| checkpointId | long | Yes | チェックポイントID | 正の整数 |
| checkpointOptions | CheckpointOptions | Yes | チェックポイントオプション | null不可 |

### 入力データソース

- JobMasterからのタスクデプロイメント命令
- 上流タスクからのデータストリーム
- チェックポイント復元データ

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| TaskExecutionState | TaskExecutionState | タスク実行状態 |
| AccumulatorSnapshot | AccumulatorSnapshot | アキュムレータ値 |
| CheckpointMetrics | CheckpointMetrics | チェックポイントメトリクス |
| TaskManagerInfo | TaskManagerInfo | TM情報 |

### 出力先

- JobMaster（状態レポート）
- 下流タスク（処理済みデータ）
- チェックポイントストレージ（状態スナップショット）

## 処理フロー

### 処理シーケンス

```
1. TaskExecutor起動
   └─ TaskManagerServicesの初期化

2. ResourceManager接続
   └─ registerTaskExecutor() でリソース登録
   └─ スロットレポート送信

3. JobMaster接続
   └─ offerSlots() でスロット提供
   └─ タスク受信準備

4. タスク実行
   └─ submitTask() でタスク受信
   └─ Task.run() でタスク実行
   └─ データ処理ループ

5. チェックポイント参加
   └─ triggerCheckpointBarrier() でバリア受信
   └─ 状態スナップショット作成
   └─ acknowledgeCheckpoint() でACK送信

6. タスク完了/終了
   └─ updateTaskExecutionState() で状態報告
   └─ リソース解放
```

### フローチャート

```mermaid
flowchart TD
    A[TaskExecutor起動] --> B[TaskManagerServices初期化]
    B --> C[ResourceManager登録]
    C --> D[ハートビート開始]

    D --> E[JobMaster接続]
    E --> F[スロット提供]
    F --> G[タスク受信]

    G --> H[Task実行]
    H --> I{処理継続?}
    I -->|Yes| J[データ処理]
    J --> K{チェックポイント?}
    K -->|Yes| L[状態スナップショット]
    L --> I
    K -->|No| I
    I -->|No| M[タスク完了]

    M --> N[状態レポート]
    N --> O[リソース解放]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | スロット管理 | 設定されたスロット数を超えてタスクを受け付けない | タスク受信時 |
| BR-002 | メモリ管理 | タスクメモリはスロット単位で分離 | タスク実行時 |
| BR-003 | ネットワークバッファ | プール化されたバッファでデータ交換 | シャッフル時 |
| BR-004 | 状態分離 | タスク間で状態を分離して管理 | ステート操作時 |

### 計算ロジック

- メモリセグメント管理（MemoryManager）
- ネットワークバッファプール管理

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

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

本機能は直接データベースを操作しない。

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TaskManagerException | システムエラー | TM起動失敗 | 設定確認・再起動 |
| SlotAllocationException | リソースエラー | スロット割当失敗 | リソース確認 |
| TaskSubmissionException | ビジネスエラー | タスクサブミット失敗 | タスク定義確認 |
| CheckpointException | 処理エラー | チェックポイント失敗 | ストレージ確認 |

### リトライ仕様

- ResourceManager接続失敗時: 定期的に再接続
- タスク障害時: JobMasterが再スケジュール

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

タスク単位でのExactly-once保証はチェックポイントによって実現。

## パフォーマンス要件

- タスク起動: 数百ミリ秒以内
- データ処理スループット: ネットワーク帯域に依存
- チェックポイント: 設定されたタイムアウト以内

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

- SSL/TLSによるタスク間通信暗号化（オプション）
- ファイアウォールによるネットワーク分離推奨

## 備考

- TaskSlotTableでスロット管理
- ShuffleEnvironmentでデータ交換管理
- FileCacheで分散ファイルキャッシュ

---

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

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

### 推奨読解順序

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

TaskExecutorで使用される主要なデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | TaskDeploymentDescriptor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java` | タスクデプロイ情報 |
| 1-2 | TaskSlot.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlot.java` | スロット定義 |
| 1-3 | UnresolvedTaskManagerLocation.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/UnresolvedTaskManagerLocation.java` | TM位置情報 |

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

TaskExecutorの起動とライフサイクルを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | TaskExecutor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java` | メインクラス |
| 2-2 | TaskManagerRunner.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java` | 起動ランナー |

**主要処理フロー**:
1. **TaskExecutor**: RpcEndpointを継承、タスク管理
2. **TaskManagerRunner**: エントリーポイント、サービス初期化

#### Step 3: タスク実行を理解する

個々のタスク実行の仕組みを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | Task.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java` | タスク実行クラス |
| 3-2 | TaskSlotTable.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTable.java` | スロット管理 |

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

```
TaskManagerRunner.main()
    │
    └─ TaskExecutor
           │
           ├─ onStart()
           │      └─ startTaskExecutorServices()
           │             ├─ ResourceManager接続
           │             └─ ハートビート開始
           │
           ├─ submitTask()
           │      ├─ TaskSlotTable.addTask()
           │      └─ Task.run()
           │             └─ StreamTask.invoke()
           │
           ├─ triggerCheckpoint()
           │      └─ Task.triggerCheckpointBarrier()
           │
           └─ freeSlot()
                  └─ TaskSlotTable.freeSlot()
```

### データフロー図

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

TaskDeployment ─────────▶ TaskExecutor ──────────────▶ TaskExecutionState
                              │
                              ▼
                         TaskSlotTable
                              │
                              ▼
                            Task
                              │
           ┌──────────────────┼──────────────────┐
           │                  │                  │
           ▼                  ▼                  ▼
    ShuffleEnvironment   MemoryManager    StateBackend
           │                  │                  │
           ▼                  ▼                  ▼
    データ交換         メモリ管理        状態管理
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| TaskExecutor.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java` | ソース | TaskExecutorメインクラス |
| TaskManagerRunner.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java` | ソース | 起動ランナー |
| Task.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java` | ソース | タスク実行 |
| TaskSlotTable.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTable.java` | ソース | スロット管理 |
| TaskManagerServices.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java` | ソース | TM共有サービス |
| TaskExecutorGateway.java | `flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java` | ソース | RPCゲートウェイ |
