Realtime Compute for Apache Flink:例外処理ガイド

1. 包括的な例外のキャッチを避ける

try {
    // ユーザーロジック
} catch (Exception e) {
    LOG.warn("Something went wrong", e);
}

try {
    processRecord(value);
} catch (JsonParseException e) {
    LOG.warn("Invalid input record: {}", value, e);
    ctx.output(ERROR_TAG, new ErrorRecord(value, e.getMessage()));
}

2. システムレベルのエラーは Flink に委任する

回復不能な状態エラー (例: 未初期化の状態、デシリアライズの失敗) が発生した場合は、例外をスローします。これにより、Flink のフェールオーバーメカニズムがトリガーされます。

if (state.value() == null) {
    // 状態が正しく初期化されていません
    throw new IllegalStateException("State not properly initialized");
}

その後、Flink は設定された再起動戦略と最後の成功したチェックポイントを使用して、ジョブを自動的に復元します。

3. 外部呼び出しのリトライを制限する

外部システム (例: データベース、HTTP サービス) とやり取りする際は、無限リトライメカニズムの実装を避けてください。

int maxRetries = 3;
for (int i = 0; i < maxRetries; i++) {
    try {
        callExternalService(record);
        return;
    } catch (IOException e) {
        if (i == maxRetries - 1) {
            throw new RuntimeException("Failed after " + maxRetries + " attempts", e);
        }
        Thread.sleep(1000 * (i + 1)); // エクスポネンシャルバックオフを使用します。
    }
}

4. 完全な例外コンテキストを保持する

e.getMessage() のみをログに記録するだけでは、トラブルシューティングには不十分なことがよくあります。代わりに、次のことを行います。

• 完全なスタックトレースをログに記録します。

• 元の入力データを保持します。

• サイド出力を使用して、ダウンストリームのモニタリングやリプレイのために構造化されたエラーイベントを出力します。

ctx.output(ERROR_TAG, new ErrorRecord(
    originalInput,
    Instant.now(),
    e.getClass().getSimpleName(),
    e.getMessage(),
    ExceptionUtils.stringifyException(e)
));

5. 例外処理パスを検証する

• 単体テストと統合テストに例外シナリオを含めます。

• コードレビュー中に、広範な catch (Exception) を使用したり、例外をサイレントに無視したりするロジックを特に精査します。

• すべての例外処理ブロックについて、「このエラーは本当に回復可能か？」、「回復可能である場合、回復ロジックは状態の整合性に影響を与えるか？」と自問します。