# 機能設計書：ブロードキャストストリーム

## 1. 機能概要

### 1.1 処理概要

ブロードキャストストリーム（BroadcastStream）は、Apache Flinkにおいて特定のデータストリームを全ての並列タスクインスタンスに配信するための機能である。この機能により、ルールデータや設定データなど、全てのパーティションで参照が必要なデータを効率的に共有できる。

本機能は2つのAPI体系を提供する：
1. **DataStream API（従来型）**: `DataStream.broadcast()` と `BroadcastConnectedStream` を使用
2. **DataStream V2 API（新型）**: `BroadcastStream` インターフェースと `TwoInputBroadcastStreamProcessFunction` を使用

### 1.2 業務上の目的・役割

| 項目 | 内容 |
|------|------|
| 目的 | 全並列タスクで同一データを共有する必要があるユースケースに対応 |
| 役割 | ルール配信、設定更新、マスタデータ同期などの実現 |
| 主な利用場面 | 動的ルールエンジン、リアルタイムアラート、フィーチャーエンリッチメント |

### 1.3 利用シーン・ユースケース

1. **動的ルール適用**
   - 不正検知ルールをリアルタイムで更新
   - 全てのトランザクション処理タスクに即時反映

2. **設定の動的変更**
   - アプリケーション設定をブロードキャスト
   - ダウンタイムなしで設定変更を反映

3. **マスタデータ結合**
   - 商品マスタ、顧客マスタ等の参照データを配信
   - ストリームデータとのエンリッチメント処理

4. **機械学習モデル配信**
   - 学習済みモデルパラメータを全タスクに配信
   - モデル更新時の即時反映

## 2. 入出力仕様

### 2.1 入力仕様

#### 2.1.1 DataStream API（従来型）

| パラメータ | 型 | 必須 | 説明 |
|-----------|-----|------|------|
| broadcastStateDescriptors | MapStateDescriptor<?, ?>... | Yes | ブロードキャストステートの記述子 |

```java
// ブロードキャストストリームの作成
MapStateDescriptor<String, Rule> ruleStateDescriptor =
    new MapStateDescriptor<>("RuleBroadcastState", String.class, Rule.class);
BroadcastStream<Rule> broadcastStream = ruleStream.broadcast(ruleStateDescriptor);
```

#### 2.1.2 DataStream V2 API（新型）

| パラメータ | 型 | 必須 | 説明 |
|-----------|-----|------|------|
| NonKeyedPartitionStream | NonKeyedPartitionStream<T_OTHER> | Yes | 非ブロードキャスト側ストリーム |
| KeyedPartitionStream | KeyedPartitionStream<K, T_OTHER> | Yes | キー付き非ブロードキャスト側ストリーム |
| processFunction | TwoInputBroadcastStreamProcessFunction | Yes | 処理関数 |
| newKeySelector | KeySelector<OUT, K> | No | 出力キーセレクタ（シャッフル回避用） |

### 2.2 出力仕様

#### 2.2.1 DataStream API

| 出力 | 型 | 説明 |
|------|-----|------|
| BroadcastStream | BroadcastStream<T> | ブロードキャストステート付きストリーム |
| BroadcastConnectedStream | BroadcastConnectedStream<IN1, IN2> | 接続済みストリーム |
| SingleOutputStreamOperator | SingleOutputStreamOperator<OUT> | 処理結果ストリーム |

#### 2.2.2 DataStream V2 API

| 出力 | 型 | 説明 |
|------|-----|------|
| NonKeyedPartitionStream | ProcessConfigurableAndNonKeyedPartitionStream<OUT> | 非キー付き出力ストリーム |
| KeyedPartitionStream | ProcessConfigurableAndKeyedPartitionStream<K, OUT> | キー付き出力ストリーム |

## 3. 処理フロー

### 3.1 全体処理フロー

```mermaid
sequenceDiagram
    participant Source as データソース
    participant Broadcast as ブロードキャストストリーム
    participant Partitioner as BroadcastPartitioner
    participant Task1 as 並列タスク1
    participant Task2 as 並列タスク2
    participant TaskN as 並列タスクN

    Source->>Broadcast: データ発行
    Broadcast->>Partitioner: パーティショニング
    Partitioner->>Task1: 全データ配信
    Partitioner->>Task2: 全データ配信
    Partitioner->>TaskN: 全データ配信

    Note over Task1,TaskN: 各タスクで同一データを処理
```

### 3.2 BroadcastConnectedStream処理フロー

```mermaid
sequenceDiagram
    participant Main as メインストリーム
    participant Broadcast as ブロードキャストストリーム
    participant Operator as BroadcastProcessOperator
    participant State as BroadcastState
    participant Output as 出力

    Broadcast->>Operator: ブロードキャストデータ到着
    Operator->>State: ステート更新（読み書き可）
    Operator->>Operator: processBroadcastElement()

    Main->>Operator: メインデータ到着
    Operator->>State: ステート参照（読み取りのみ）
    Operator->>Operator: processElement()
    Operator->>Output: 結果出力
```

### 3.3 DataStream V2 API処理フロー

```mermaid
flowchart TD
    subgraph Input["入力ストリーム"]
        BS[BroadcastStream]
        NS[NonKeyedPartitionStream]
        KS[KeyedPartitionStream]
    end

    subgraph Processing["処理レイヤー"]
        BI[BroadcastStreamImpl]
        TOP[TwoInputBroadcastProcessOperator]
        KTOP[KeyedTwoInputBroadcastProcessOperator]
        PF[TwoInputBroadcastStreamProcessFunction]
    end

    subgraph Output["出力ストリーム"]
        NOS[NonKeyedPartitionStream]
        KOS[KeyedPartitionStream]
    end

    BS --> BI
    BI --> TOP
    BI --> KTOP
    NS --> TOP
    KS --> KTOP

    TOP --> PF
    KTOP --> PF

    PF --> NOS
    PF --> KOS
```

## 4. 業務ルール・条件

### 4.1 ブロードキャストステートアクセスルール

| 入力側 | ステートアクセス | 説明 |
|--------|----------------|------|
| ブロードキャスト側 | 読み書き可 | Context経由でBroadcastStateを更新可能 |
| 非ブロードキャスト側 | 読み取りのみ | ReadOnlyContext経由でReadOnlyBroadcastStateのみアクセス可能 |

### 4.2 ストリーム接続パターン

| パターン | 非ブロードキャスト側 | 処理関数 | 備考 |
|---------|---------------------|----------|------|
| 非キー付き接続 | DataStream | BroadcastProcessFunction | ステート再配分不可 |
| キー付き接続 | KeyedStream | KeyedBroadcastProcessFunction | キー付きステートとブロードキャストステートを併用可能 |

### 4.3 DataStream V2 APIでのステート制約

```java
// NonKeyedPartitionStreamとの接続時のステート検証
validateStates(
    processFunction.usesStates(),
    new HashSet<>(Collections.singletonList(StateDeclaration.RedistributionMode.NONE)));
```

- **RedistributionMode.NONE**: 非キー付きストリームではステート再配分モードはNONEのみ許可

### 4.4 キー保持パターン

```java
// シャッフル回避のためのキー保持
<K, T_OTHER, OUT> ProcessConfigurableAndKeyedPartitionStream<K, OUT> connectAndProcess(
    KeyedPartitionStream<K, T_OTHER> other,
    TwoInputBroadcastStreamProcessFunction<T_OTHER, T, OUT> processFunction,
    KeySelector<OUT, K> newKeySelector);
```

- 新しいKeySelectorは元のKeySelectorと同じキーを抽出する必要がある
- ブロードキャスト入力からの出力はキーパーティション自体から出力キーを取得

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

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

```
DataStream API（従来型）
├── DataStream.broadcast(MapStateDescriptor...)
│   └── BroadcastStream
│       └── DataStream.connect(BroadcastStream)
│           └── BroadcastConnectedStream
│               ├── process(BroadcastProcessFunction)
│               │   └── BroadcastStateTransformation
│               └── process(KeyedBroadcastProcessFunction)
│                   └── KeyedBroadcastStateTransformation

DataStream V2 API
├── BroadcastStream (interface)
│   └── BroadcastStreamImpl
│       ├── connectAndProcess(KeyedPartitionStream, TwoInputBroadcastStreamProcessFunction)
│       │   └── KeyedTwoInputBroadcastProcessOperator
│       ├── connectAndProcess(NonKeyedPartitionStream, TwoInputBroadcastStreamProcessFunction)
│       │   └── TwoInputBroadcastProcessOperator
│       └── connectAndProcess(..., KeySelector) [シャッフル回避版]
│           └── KeyedTwoInputBroadcastProcessOperator

共通基盤
└── BroadcastPartitioner
    └── isBroadcast() -> true
```

### 5.2 データフロー図

```
[DataSource] ─────────────────────────────────────────────────────────┐
     │                                                                 │
     ▼                                                                 │
[DataStream.broadcast()] ───────────────────────────────────────────┐ │
     │                                                               │ │
     ▼                                                               │ │
[BroadcastPartitioner] ─── selectChannel() → UnsupportedOperation   │ │
     │                     isBroadcast() → true                      │ │
     │                                                               │ │
     ▼                                                               │ │
[全並列タスクに複製配信]                                              │ │
     │                                                               │ │
     └──────────────────┬──────────────────────────────────────────────┘
                        ▼
              [BroadcastConnectedStream]
                        │
          ┌─────────────┴─────────────┐
          ▼                           ▼
    [processElement]           [processBroadcastElement]
    (ReadOnlyContext)          (Context - 読み書き可)
          │                           │
          └───────────┬───────────────┘
                      ▼
              [BroadcastState]
                      │
                      ▼
                  [Output]
```

### 5.3 関連ファイル一覧

| ファイルパス | 役割 |
|-------------|------|
| `flink-datastream-api/.../BroadcastStream.java` | DataStream V2 BroadcastStreamインターフェース |
| `flink-datastream/.../BroadcastStreamImpl.java` | DataStream V2 BroadcastStream実装 |
| `flink-datastream-api/.../TwoInputBroadcastStreamProcessFunction.java` | V2 API処理関数インターフェース |
| `flink-datastream/.../TwoInputBroadcastProcessOperator.java` | V2 API非キー付きオペレータ |
| `flink-datastream/.../KeyedTwoInputBroadcastProcessOperator.java` | V2 APIキー付きオペレータ |
| `flink-runtime/.../BroadcastStream.java` | DataStream API BroadcastStreamクラス |
| `flink-runtime/.../BroadcastConnectedStream.java` | DataStream API接続済みストリーム |
| `flink-runtime/.../BroadcastProcessFunction.java` | 非キー付き処理関数 |
| `flink-runtime/.../BaseBroadcastProcessFunction.java` | 処理関数基底クラス |
| `flink-runtime/.../BroadcastPartitioner.java` | ブロードキャストパーティショナー |

### 5.4 コードリーディング手順

#### Step 1: BroadcastPartitionerの理解（基盤層）

```java
// flink-runtime/.../BroadcastPartitioner.java (31-73行目)
@Internal
public class BroadcastPartitioner<T> extends StreamPartitioner<T> {
    @Override
    public int selectChannel(SerializationDelegate<StreamRecord<T>> record) {
        throw new UnsupportedOperationException(
                "Broadcast partitioner does not support select channels.");
    }

    @Override
    public boolean isBroadcast() {
        return true;
    }
}
```

- `selectChannel()` は呼び出されない（全チャネルに配信するため）
- `isBroadcast()` が `true` を返すことで、RecordWriterが全チャネルに書き込む

#### Step 2: DataStream APIエントリーポイント

```java
// flink-runtime/.../DataStream.java (339-344行目)
public BroadcastStream<T> broadcast(
        final MapStateDescriptor<?, ?>... broadcastStateDescriptors) {
    Preconditions.checkNotNull(broadcastStateDescriptors);
    final DataStream<T> broadcastStream = setConnectionType(new BroadcastPartitioner<>());
    return new BroadcastStream<>(environment, broadcastStream, broadcastStateDescriptors);
}
```

#### Step 3: 接続と処理関数の適用

```java
// flink-runtime/.../BroadcastConnectedStream.java (178-219行目)
public <OUT> SingleOutputStreamOperator<OUT> process(
        final BroadcastProcessFunction<IN1, IN2, OUT> function) {
    // ...
    return transform(function, outTypeInfo);
}
```

#### Step 4: DataStream V2 APIの処理フロー

```java
// flink-datastream/.../BroadcastStreamImpl.java (59-90行目)
public <K, T_OTHER, OUT> ProcessConfigurableAndNonKeyedPartitionStream<OUT> connectAndProcess(
        KeyedPartitionStream<K, T_OTHER> other,
        TwoInputBroadcastStreamProcessFunction<T_OTHER, T, OUT> processFunction) {
    // ブロードキャスト入力は常に第2入力として扱う
    KeyedTwoInputBroadcastProcessOperator<K, T_OTHER, T, OUT> processOperator =
            new KeyedTwoInputBroadcastProcessOperator<>(processFunction);
    // ...
}
```

#### Step 5: 処理関数インターフェースの理解

```java
// flink-datastream-api/.../TwoInputBroadcastStreamProcessFunction.java (53-65行目)
void processRecordFromNonBroadcastInput(
        IN1 record, Collector<OUT> output, PartitionedContext<OUT> ctx) throws Exception;

void processRecordFromBroadcastInput(IN2 record, NonPartitionedContext<OUT> ctx)
        throws Exception;
```

- ブロードキャスト側は `NonPartitionedContext` を受け取る（全パーティション対応のため）
- 非ブロードキャスト側は `PartitionedContext` を受け取る（個別パーティション処理）

## 6. 関連機能・API

### 6.1 関連機能

| 機能名 | 関連種別 | 説明 |
|--------|---------|------|
| DataStream変換（No.7） | 前提機能 | ストリーム変換の基盤 |
| Watermark（No.10） | 連携機能 | イベント時間処理との連携 |
| ステート管理（No.27） | 連携機能 | BroadcastStateの管理 |
| チェックポイント（No.26） | 連携機能 | ブロードキャストステートの永続化 |

### 6.2 APIリファレンス

#### DataStream API

```java
// ブロードキャストストリーム作成
public BroadcastStream<T> broadcast(MapStateDescriptor<?, ?>... broadcastStateDescriptors)

// 接続
public <R> BroadcastConnectedStream<T, R> connect(BroadcastStream<R> broadcastStream)

// 処理（非キー付き）
public <OUT> SingleOutputStreamOperator<OUT> process(BroadcastProcessFunction<IN1, IN2, OUT> function)

// 処理（キー付き）
public <KEY, OUT> SingleOutputStreamOperator<OUT> process(KeyedBroadcastProcessFunction<KEY, IN1, IN2, OUT> function)
```

#### DataStream V2 API

```java
// BroadcastStreamインターフェース
public interface BroadcastStream<T> extends DataStream {
    <K, T_OTHER, OUT> ProcessConfigurableAndNonKeyedPartitionStream<OUT> connectAndProcess(
            KeyedPartitionStream<K, T_OTHER> other,
            TwoInputBroadcastStreamProcessFunction<T_OTHER, T, OUT> processFunction);

    <T_OTHER, OUT> ProcessConfigurableAndNonKeyedPartitionStream<OUT> connectAndProcess(
            NonKeyedPartitionStream<T_OTHER> other,
            TwoInputBroadcastStreamProcessFunction<T_OTHER, T, OUT> processFunction);

    <K, T_OTHER, OUT> ProcessConfigurableAndKeyedPartitionStream<K, OUT> connectAndProcess(
            KeyedPartitionStream<K, T_OTHER> other,
            TwoInputBroadcastStreamProcessFunction<T_OTHER, T, OUT> processFunction,
            KeySelector<OUT, K> newKeySelector);
}
```

## 7. 設計上の考慮事項

### 7.1 パフォーマンス考慮事項

| 考慮点 | 説明 | 推奨対応 |
|--------|------|---------|
| メモリ使用量 | 全タスクでブロードキャストデータを保持 | ブロードキャストデータは小さく保つ |
| ネットワーク負荷 | 全タスクへの配信でN倍の通信量 | 頻繁な更新を避ける |
| ステート同期 | 並列タスク間でのステート一貫性 | 冪等性のある更新処理を実装 |

### 7.2 ステート管理の注意点

1. **ブロードキャストステートの一貫性**
   - 全並列タスクで同じ順序でブロードキャストデータが到着するとは限らない
   - 非ブロードキャスト入力との順序も保証されない

2. **アクセス権限の厳格な分離**
   - ブロードキャスト側: 読み書き可（`BroadcastState`）
   - 非ブロードキャスト側: 読み取りのみ（`ReadOnlyBroadcastState`）

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

| 観点 | リスク | 対策 |
|------|--------|------|
| データ配信 | 機密データの全タスク配信 | 必要最小限のデータのみブロードキャスト |
| ステートアクセス | 意図しないステート変更 | ReadOnlyContextによるアクセス制御 |

### 7.4 障害復旧

- ブロードキャストステートはチェックポイントに含まれる
- 復旧時は全タスクで同一のステートが復元される
- `SubtaskStateMapper.UNSUPPORTED` により、並列度変更時のステート再配分は未サポート

## 8. 用語集

| 用語 | 説明 |
|------|------|
| BroadcastStream | 全並列タスクにデータを配信するストリーム |
| BroadcastState | ブロードキャストストリームと関連付けられたキー・バリュー形式のステート |
| BroadcastConnectedStream | ブロードキャストストリームと通常ストリームを接続した結果 |
| BroadcastPartitioner | 全チャネルにデータを配信するパーティショナー |
| MapStateDescriptor | ブロードキャストステートの型情報を定義する記述子 |
| ReadOnlyBroadcastState | 読み取り専用のブロードキャストステートビュー |
| NonPartitionedContext | パーティション横断的な処理コンテキスト |
| PartitionedContext | 個別パーティション処理コンテキスト |
