[chikamura]

Summary

The java port swallowed two data-loss errors that ruby raises: a record conversion/write error truncated the intermediate file without failing the run, and an upload IOException was logged at INFO and ignored.

Data phase

• BigqueryPageOutput#add() and BigqueryFileWriter#write() no longer swallow errors; the run fails at once.
• BigqueryFileWriter#close() still swallows (ruby: io.close rescue nil) but logs at WARN, creates no file for a writer that never wrote, and always releases the file descriptor (GZIP leaked it when the trailer or the header write failed).
• transaction() loads the files this run's writers wrote instead of globbing path_prefix*file_ext. The cleanup runs every step and closes every writer even if one throws, and keeps the run's own failure as the reported one, with cleanup failures attached as suppressed.

Upload (BigqueryClient#load())

• The job id is fixed for the whole upload, like ruby's with_network_retry, so a retry never creates a second job.
• The library retries each request inside the session; load() re-sends the whole file only for a socket error under TLS, which the library doesn't retry. Re-sending on top of the library's attempts multiplied them by retries.
• The write channel is closed only after the whole file is written, so a failed attempt no longer finalizes a partial upload into a load job.
• A 409 on the retry waits for the existing job instead of failing; the lookup carries the job's location (location, else the dataset's). After any other transient failure, the upload asks for its job id once before failing and waits for the job if it exists.

Behavior change

• location, when configured, now goes into every job reference (as ruby does), so a location that does not match the dataset's is rejected by BigQuery where before it was ignored.

Differences from ruby, on purpose

• A failed run's partial file is cleaned up; a job a failed upload left behind is waited for; a cleanup failure doesn't replace the run's failure; a failed close is logged; an empty page leaves no file.

Follow-up (separate PR)

• Clearing BigqueryUtil.getFileWriters() between two transaction() calls in one JVM (same shape in ruby; the NPE part is ruby-compat/14).

Test plan

• ./gradlew test (188 tests, 1 skipped)
• ./gradlew spotlessApply

🤖 Generated with Claude Code

[john-abre]

ruby-compat/07（親 issue の PR7 = 2-4, 2-5）のレビューです。方向性は分割案どおりで、BigqueryPageOutput#add() の握りつぶし除去と BigqueryFileWriter#write() の再スローは問題ありません。ただし、目玉であるアップロード周りのリトライ経路に 2 点、修正が必要な問題があります。いずれもライブラリのバイトコード（google-cloud-bigquery 2.14.0 / google-cloud-core 2.8.0）で確認済みです。

修正必須（インラインコメント参照）

1. 実際のネットワーク障害はリトライされません。 新しい catch (IOException) は、書き込みチャネルが投げる BigQueryException（RuntimeException）を捕まえられないため、接続リセットなどは 1 回目で即 give up になります。
2. リトライで行が二重ロードされる可能性があります。 失敗後も try-with-resources がチャネルを close し、その close が部分アップロードを finalize してロードジョブを開始します。その後のリトライは新しい job id で完全なファイルをロードするため、append_direct では両方が追記されます。ruby は job_id を 1 回だけ生成してリトライ間で使い回し、まさにこの重複を防いでいます（bigquery_client.rb のコメント参照）。

親 issue の文脈について

• 分割案では 2-5 は「ERROR ログ + 即時失敗」であり、リトライの追加は計画外です。追加するなら ruby と同じく job_id をリトライループの外で生成する必要がありますし、計画どおり即時失敗にするのも選択肢です。
• 後続で予定されている 429 リトライのラッパー（bigquery.* 呼び出しを包むもの）では 1. は解決しません。障害は bigquery.writer() ではなくストリームの write / close から出るためです。この PR で直す必要があります。
• テストで実際に 2 秒スリープしている点は、#113 の後に rebase して buildJobRetryExecutor() に乗せれば（build.gradle の環境変数で 1ms になるため）解消します。rebase 時の注意点としてだけ挙げておきます。

修正推奨: close() の flush 失敗時のストリームリーク、write() の素の RuntimeException。

軽微: スタックトレースの重複出力、BUFFER_SIZE の public 化。中間ファイルが存在しない場合の return null は既存挙動で 2-4 / 2-5 の範囲外なので、issue にメモして別 PR でよいと思います。

[john-abre]

修正必須: この catch (IOException) は、実際のネットワーク障害を捕まえられません。

Channels.newOutputStream(writer).write() → BaseWriteChannel.write() → TableDataWriteChannel.flushBuffer() という経路で、flushBuffer は RetryHelperException を catch して BigQueryException.translateAndThrow で BigQueryException（RuntimeException） に変換して投げます（2.14.0 のバイトコードで確認）。そのため接続リセット等はこの catch を素通りし、isRetryableException() も false を返すので、1 回目で「Give up retrying」となり retries の回数に関わらずリトライされません。PR 説明の「ruby の with_network_retry と同様に retries 回リトライ」は現状成り立っていません。

逆に、この catch に入る IOException は Files.copy のローカル側読み込みエラー（ファイル消失・読み取り不可など）だけで、これらは一過性ではないため 2〜10 秒のバックオフ付きリトライは無駄になります。

アップロードと close を囲む範囲で BigQueryException も catch（必要なら isRetryable() で絞る）して BigqueryUploadException に包んでください。なお、後続で予定されている 429 リトライのラッパーは bigquery.* の呼び出し（writer() など）を包むものなので、ストリームの write / close で発生するこの例外はカバーされません。この PR で直す必要があります。

[john-abre]

修正必須: アップロード途中で失敗しても、try-with-resources がチャネルを close します。BaseWriteChannel.close() は isOpen なら flushBuffer(length, lastChunk=true) を呼ぶ（バイトコードで確認）ので、部分的なバイト列で resumable セッションが finalize され、BigQuery 側で部分ファイルのロードジョブ（job id = uuid1）が開始されます。その後 BigqueryUploadException で RetryExecutor がリトライし、call() が再実行されて新しい uuid2 で完全なファイルがロードされます。

append_direct モードでは両方のジョブが宛先テーブルに追記され、num_output_rows は成功した uuid2 の LoadStatistics しか合計しないため、行数チェックでは uuid1 の重複行を検出できません。この PR 以前は握りつぶした後に 1 つの部分ジョブを waitForLoad() していたので、少なくとも 2 つ目のロードは発行されませんでした。

ruby 版はこの点を明示的に防いでいます: load() で job_id を リトライの外で 1 回だけ 生成し（# we should generate job_id in client code, otherwise, retrying would cause duplication）、with_network_retry の中で同じ job_id を使い回すので、リトライ時に重複ジョブが作られることはありません。

対応案: (a) ruby と同じく uuid / jobId の生成をリトライループ（call()）の外に出し、リトライでは同じ job id を再利用する（finalize 済みなら BigQuery が重複 job id を拒否するので、重複ロードではなく失敗になります）。(b) 分割案（2-5 は「ERROR ログ + 即時失敗」）どおりアップロードのリトライ自体をやめる。いずれにせよ失敗後にチャネルを close して finalize してしまう点は避ける必要があります。

[john-abre]

上のコメントに関連: jobId が call() の中で生成されているため、リトライのたびに新しいジョブになります。ruby は load() 1 回につき job_id を 1 つだけ生成し、リトライ間で使い回しています。リトライを残すなら、ここをループの外に出すのが ruby との整合という意味でも正しい形だと思います。

[john-abre]

軽微: ここでリトライ本体の中から毎回フルスタックトレース付きで ERROR を出し、さらに onRetry が retries 回ごとに WARN でスタック付き、最後に例外が伝播して Embulk も出力するので、retries: 5 で障害が継続すると同じトレースが最大 8 回出てログに埋もれます。ループ内はメッセージのみ（throwable なし）にして、スタックは伝播する例外に任せるのがよさそうです。

[john-abre]

修正推奨: flush() と close() が 1 つの try に入っているため、flush() が失敗すると close() が呼ばれず、FileOutputStream がリークします（GZIP の場合はトレーラーも書かれません）。

またコメントの「行数チェックで後から検出される」は常には成り立ちません。abort_on_error のデフォルトは max_bad_records == 0 なので、max_bad_records > 0 か is_skip_job_result_check: true の設定では num_input_rows / num_output_rows の比較がスキップされ、末尾の最大 32KB 分のレコードが欠けたままジョブが成功します。まさにこの PR が塞ごうとしているデータ欠損のクラスです。close() は finally（または try-with-resources）で必ず呼び、ログは少なくとも WARN / ERROR にしてください。

[john-abre]

修正推奨: 素の RuntimeException(e) ではなく、この PR で追加した BigqueryException(String, Throwable) を使ってください。現状だと Embulk の PartialExecutionException には java.lang.RuntimeException: java.io.IOException: No space left on device のようにプラグインの文脈なしで出ますし、ログにしかない「failed to write an intermediate file」のメッセージが例外に乗りません。モジュール内の他の失敗（BigqueryClient / BigqueryJobWaiter）はすべて BigqueryException 系なので、型も揃います。

[john-abre]

nit: BUFFER_SIZE は同一パッケージの TestBigqueryFileWriter から参照するためだけに public になっています。package-private で十分なので、プラグインの公開 API を広げないほうがよいと思います。

[john-abre]

rebase 時の注意: この fixture で動く 2 つのリトライテストは実際の RetryExecutor（初期待機 2 秒）を回すので、./gradlew test 全体に約 4 秒の実スリープが加わります。親 issue の着手順では #113 の後に rebase する想定なので、その際に load() のインラインの RetryExecutor を #113 の buildJobRetryExecutor() に乗せれば、build.gradle の BIGQUERY_OUTPUT_OPTION_JOB_RETRY_* で待機が 1ms になり解消します。この PR 単体で直す必要はありません。

[john-abre]

ありがとうございます。変換エラーを握りつぶさずに即座に失敗させる方針は正しく、同一 ID でのアップロードリトライも ruby の with_network_retry と一致しています。マージ前に修正が必要な点が 2 つあり、どちらもこの PR によって新たに表面化するデータ整合性の問題です。それ以外は任意です。

要修正

1. 失敗した実行で中途半端な中間ファイルが残る。 add() が例外を投げるようになったため、BigqueryJavaOutputPlugin.transaction() 内の control.run() から例外が抜けます。writer の close と getIntermediateFiles() はどちらもその呼び出しの後にあり、finally は paths != null の場合にしか削除しないため、書きかけの <path_prefix>.<pid>.<tid>.jsonl が開いたままディスク上に残ります。getIntermediateFiles は path_prefix*file_ext で glob するので、次回の実行で古いファイルも自分のファイルと一緒に投入され、途中までの行がロードされてしまいます。abort_on_error は行数の不一致をロード後にしか検知しないため、一時テーブルを使うモードは守られますが append_direct は守られません。修正案：finally 内で、paths が設定されたかどうかに関わらず、登録済みのすべての writer を close し、delete_from_local_when_job_end が有効ならそのファイルを削除する。

2. 同一ジョブ ID でのリトライがリトライ不可の 409 で失敗し得る。 uploadOnce では stream.close() が最後のチャンクを送信し、BigQuery がセッションを確定して embulk_load_job_<uuid> を作成します。そのリクエストのレスポンスが失われた場合（connection reset）、例外はリトライ可能と判定されるため upload() は同じ ID で再送します。2 回目の確定では 409 Already Exists: Job が返り、isRetryable() は false なので、最初のロードジョブが存在して実行されているにも関わらず実行全体が失敗します。append_direct では行が追加済みの状態となり、ユーザーが再実行すると重複が発生します。修正案：同一ジョブ ID での 409 を成功として扱い、bigquery.getJob(JobId.of(jobId)) で取得したジョブを waitForLoad に渡す。取得したジョブが失敗していた場合は、外側のジョブリトライが新しい ID で処理するので問題ありません。

任意のコメント

• BigqueryFileWriter.close() は outputStream() を呼び出しており、これがファイルを遅延オープンします。setFileWriter で登録されたがレコードを一件も書かなかった writer（空ページ）は 0 バイトのファイルを残し、同じ glob によってロードジョブとして投入されます。既存の問題ですが、今回触るついでに if (os == null) return; のガードを入れる価値があります。
• uploadOnce 内の IOException は（コメントの通り）ローカルファイルの読み込みからしか発生しませんが、BigqueryUploadException に変換されて retries 回リトライされます。ruby はソケット／接続エラーのみリトライするので、ファイルが存在しない・読めない場合は即座に失敗させるべきです。
• TLS レベルのリセット（javax.net.ssl.SSLException: Connection reset）は google-cloud-core 2.8.0 では isRetryable() にならないため、新しいリトライを素通りします。ruby でもリトライされないのでパリティの破綻ではなく、目的に対するギャップですが、ruby のようなメッセージベースのチェックでカバーできます。
• 失敗した各試行が uploadOnce で ERROR として、さらに upload() で WARN/ERROR としてログ出力されるため、一時的なエラーから回復しても ERROR 行が残り、リトライを使い切った場合は同じ文言が 3 回出力されます。ruby はリトライ時に WARN、諦めた時にのみ ERROR を出力するので、upload() で一度だけログを出せば一致します。
• load() はすでに RetryExecutor を使っているのに、upload() は独自のリトライループを実装しています。RetryExecutor.builder().withRetryLimit(task.getRetries()).withInitialRetryWaitMillis(0).withMaxRetryWaitMillis(0) に isRetryableException = e instanceof BigqueryUploadException を組み合わせれば、同じ書き方で即時リトライを再現できます。
• BigqueryFileWriter.close() の try-with-resources 内での明示的な stream.flush() は冗長です。BufferedOutputStream／GZIPOutputStream の close() がすでに flush します。

🤖 Generated with Claude Code

[john-abre]

高精度レビュー(recall 重視)で 8 件のインラインコメントを付けました。重要度順: 409 救済パスの location 欠落(高)、GZIP close 時のリーク(中)、409 後の getJob 例外の取りこぼし(中)、リトライの掛け算(中/効率)、finally での例外上書き(低〜中)、テストの static マップ汚染(低)、dead な null チェック(低)、リトライ展開の重複(低)。

---

追記 (10/16): 投稿時の手違いでインラインコメントの本文が空になっていたため、削除して個別に投稿し直しました。409 救済パス関連の 2 件(location 欠落、getJob 例外の取りこぼし)は 0cc6780 で対応済みと確認できたので再投稿していません。残りの 6 件に加え、BigqueryUtil.getFileWriters() が transaction 間でクリアされない件を 1 件追加しています。

[john-abre]

BigqueryUtil.getFileWriters() は JVM 全体で共有される static な map ですが、transaction() の開始時にも終了時にもクリアされておらず、BigqueryFileWriter.close() も os / path をリセットしません。このため、同一 JVM 内で transaction() が 2 回以上呼ばれる場合（embedded Embulk や、プロセスを使い回すホスト）に、前回の run の writer がそのまま残ります。

• delete_from_local_when_job_end: true の場合: 前回の path はもう存在しないので load() が「does not exist, skipped」分岐で null を返し、statistics.add(null) → getTransactionReport の LoadStatistics::getOutputRows で NPE になります。
• false の場合: 前回のファイルがもう一度ロードされ、行が重複します。
• また、ThreadLocal の writer を再利用したスレッドは close 済みの stream に書き込むことになり、write() が「Stream Closed」で失敗します。

base ブランチでも map 自体はリークしていましたが、そこでは行数カウントにしか使われていませんでした。この PR でロード対象ファイルの決定にも使われるようになったので影響範囲が広がっています（テストは @Before で getFileWriters().clear() しているため通っています）。

transaction() の先頭で map をクリアする（あわせて writer の状態もリセットする）のが根本的な対処になると思います。

[john-abre]

[重要度: 中] GZIP の場合、close() 中の flush 失敗で FileOutputStream と Deflater がリークする(コメントの前提が成り立たない)

コメントでは「DeflaterOutputStream は flush が失敗しても下層のファイルを閉じる」とありますが、DeflaterOutputStream#close() は

if (!closed) { finish(); if (usesDefaultDeflater) def.end(); out.close(); closed = true; }

という実装で、finish()(gzip トレーラの書き込み)が IOException を投げると out.close() も def.end() も実行されません。ディスクフルなどで close 時に失敗すると、compression: GZIP のときは FileOutputStream の fd とネイティブの Deflater が GC までリークします(finally 側の delete() 自体は POSIX では通りますが、ハンドルは開いたままです)。

testCloseClosesStreamEvenIfFlushFails は自作の OutputStream で検証しているため、この GZIP 経路はカバーされていません。open() で生の FileOutputStream を別フィールドに保持しておき、close() の finally でそれも閉じる、あるいは GZIPOutputStream の close 失敗時に明示的に下層を閉じる形にすると安全です。

[john-abre]

[重要度: 中 / 効率] ライブラリのリトライと upload() のリトライが掛け算になり、最悪で (retries+1)² 回の送信と retries+1 倍の total timeout になる

buildRetrySettings() でライブラリ側に maxAttempts = retries + 1(+ バックオフ、retry_total_timeout_sec 上限)を設定した上で、upload() がセッションを開き直してさらに retries 回やり直します。新しいセッションではライブラリのバックオフ予算もリセットされるので、「ライブラリのバックオフはもう消化済み」というコメントは各セッション単位でしか成り立たず、retries: 5(デフォルト)で恒常的に 5xx を返すエンドポイントに対しては 6 セッション × 6 回 = 36 回の PUT、待ち時間は最大で retry_total_timeout_sec の 6 倍になります。大きなファイルでは毎回フルで再送するので転送量も同様に増えます。

ruby 側は google-api-client のリトライが 0 で with_network_retry が唯一のリトライ層なので、こちらの方がかなり多くリトライしています。upload() 側を TLS 配下のソケットエラー(ライブラリが拾わないもの)と 409 の救済に限定し、ライブラリが既に retries 回試した失敗(e.isRetryable() が真のもの)は外側では再送しない、あるいは外側の回数を別設定にする、のいずれかで揃える方が意図に近いように思います。

[john-abre]

[重要度: 低〜中] finally 内の close() / deleteTable() が投げた例外が、本来の失敗原因(データ変換エラーなど)を上書きする

Java では try 側の例外が伝播中に finally 側で別の例外が投げられると、元の例外は捨てられます。ここでは control.run() が「hello cannot be converted to INTEGER」で失敗した後、finally の close()(RuntimeException の場合)や deleteTable()(BigQuery のネットワークエラー)が投げると、ユーザーに見えるのは後者だけになります。deleteTable については以前からの挙動ですが、今回 close の層が一つ増えました。testRunStillCleansUpWhenClosingAWriterThrows は最初の close と finally の close が同じメッセージを投げるため、この上書きを検出できていません。

finally 内の各ステップは try/catch で握ってログに出す(または元の例外に addSuppressed する)形にして、元の失敗原因が必ず伝播するようにすると良さそうです。

[john-abre]

[重要度: 低 / テスト] 常に例外を投げる writer を JVM 全体の static マップに入れたまま削除していない

brokenWriter は close() のたびに RuntimeException を投げる実装で、-1L キーで BigqueryUtil.getFileWriters() に残り続けます。現状はこのクラスの @Before clearFileWriters() が次のテストの前に消してくれるので事故になりませんが、同じ JVM で後に実行される別クラスが transaction() を呼ぶ(あるいは @Before が外れる)と、無関係なテストが boom: close failed で落ちます。

try { ... } finally { BigqueryUtil.getFileWriters().remove(-1L); } のようにこのテスト自身で片付けるのが安全です。

[john-abre]

[重要度: 低 / 簡素化] if (paths != null) は常に真になった

直前で paths == null のときに getIntermediateFiles(...) を代入しており、この関数は Collectors.toList() の結果を返すので null を返しません。旧実装(task 版)の名残の null チェックなので、外して一段ネストを減らせます。

[john-abre]

[重要度: 低 / 重複] RetryGiveupException / InterruptedException の展開処理が 6 箇所目になっている

load()・copy()・runQuery()・executeQuery() 等と同じ「cause が BigqueryException ならそれを投げ直し、そうでなければ RuntimeException に包み、InterruptedException は BigqueryException("interrupted")」というブロックがここにも複製されています。private <T> T runWithRetry(RetryExecutor, Retryable<T>) のようなヘルパーに寄せると、今回のように onGiveup で lastException を投げ直す流儀も一箇所で揃えられます。

[chikamura]

@john-abre
Thanks for the second round. Addressed as follows.

GZIP close leak
Fixed (2095446, f57540d). The writer keeps the FileOutputStream it opened and closes it after closing the stream, whatever that did; open() releases it itself when wrapping it fails. The comment claiming the stream closed the file on its own was wrong for GZIP.

Retry multiplication
Fixed (d4516d2). upload() re-sends the file only for what the library never retries (a socket error under TLS); what it does retry it has already re-sent inside the session. Since that removes the resend that used to run into the 409, the upload now asks for its job id once after a transient failure and waits for the job if it exists (db6021a, 32062ec, 312ffac).

Exception overwrite in finally
Fixed (6413745, 2dc15d4, c8be5cd, f78dbcf, 6135929). A cleanup failure (a writer that can't be closed, a temp table that can't be deleted) is attached to the run's own failure as suppressed, and logged; with no such failure it still fails the run. Every writer is closed even if one throws.

Static map in the test
Fixed (01485d0). The test removes the writer in a finally.

Dead null check
Removed (5c76b93).

File-writer map not cleared between transactions
Not in this PR. ruby has the same shape (@file_writers is never reset in transaction either). The NPE part is the load() null for a missing file, which ruby-compat/14 turns into ruby's 0-row report; the map clear and writer reset will come in a separate PR with a test that runs two transactions in one JVM.