Stream has been finalized and cannot be appended during Flink job recovery after task manager restart

[lpvcpp]

Description

When running a Flink streaming job using the BigQuery Sink with DeliveryGuarantee.EXACTLY_ONCE, the job fails during recovery if a task manager (pod) shuts down and restarts. The connector attempts to append to a BigQuery stream that has already been finalized, resulting in an INVALID_ARGUMENT exception. This leads to repeated job failures after restarts.

This issue occurs in a distributed setup (e.g., 4 task managers), where Flink restores state from checkpoints, but the stored stream ID is no longer appendable. The connector does not automatically create a new stream during recovery, causing the commit to fail.

Steps to Reproduce

1. Set up a Flink job with the BigQuery Sink:
   • Use BigQuerySinkConfig with DeliveryGuarantee.EXACTLY_ONCE.
   • Parallelism >1 (e.g., 4 task managers).
   • Sink writes to a partitioned table (e.g., on "etl_date" field).
   • Example config snippet:

     DeliveryGuarantee deliveryGuarantee = DeliveryGuarantee.EXACTLY_ONCE;
     BigQuerySinkConfig<GenericRecord> sinkConfig = BigQuerySinkConfig.<GenericRecord>newBuilder()
         .connectOptions(BigQueryConnectOptions.builder()
             .project("xxx")
             .dataset("xxx")
             .table("xxx")
             .build())
         .deliveryGuarantee(deliveryGuarantee)
         .serializer(new AvroToProtoSerializer())
         .enableTableCreation(true)
         .partitionField("etl_date")
         .partitionExpirationMillis(45L * 24L * 60L * 60L * 1000L)
         .build();

2. Run the job in a cluster (e.g., Google Cloud Dataproc or Kubernetes) processing unbounded data (e.g., from Kafka).
3. Simulate a failure: Shut down one task manager pod (e.g., kill pod in K8s).
4. Flink attempts to recover from checkpoint, but the commit fails with the error.

Environment

• Flink version: 1.19.1
• Connector version: 1.11
• Java version: 17
• Deployment: Kubernetes
• BigQuery details: Writing to table (time-partitioned on "etl_date" as string "yyyy-MM-dd").
• Checkpoint interval: [e.g., 1 minute]

[seokjin0414]

We hit this failure in production (connector 1.1.0, Flink 1.20, exactly-once sink, Flink Kubernetes Operator, 5 sink subtasks) and traced it to the point where it can be reproduced and fixed. Sharing the analysis since the issue has no root-cause discussion yet. The mechanism is not in the writer's append path but in how the writer finalizes streams that the committer still needs.

Root cause: the writer finalizes the stream that checkpoint state still references, and BigQuery then rejects the committer's FlushRows on it.

Two facts combine:

1. BigQuery rejects FlushRows on a finalized BUFFERED stream even for an offset that was already flushed. Verified with the storage client against a scratch table: after FinalizeWriteStream, FlushRows(offset already flushed) and FlushRows(unflushed tail) both fail with INVALID_ARGUMENT: Stream has been finalized and cannot be appended: ... (a gax InvalidArgumentException wrapping the StatusRuntimeException). Re-flushing the same offset on an open stream is idempotent and succeeds.
2. Flink's CommitterOperator re-commits the restored committables synchronously in initializeState ("try to re-commit recovered transactions as quickly as possible"). BigQueryCommitter.commit turns any ApiException from flushRows into BigQueryConnectorException("Commit operation failed").

BigQueryBufferedWriter finalizes the stream held in checkpoint state in two places: discardStream() (first append after a restore rejected with OffsetAlreadyExists / OffsetOutOfRange / StreamFinalizedException / StreamNotFound, or timed out) and close() whenever the writer appended anything after the last snapshot. After that, the checkpointed committable (stream, offset) can never be flushed again.

Sequence in our incident (timestamps UTC):

• 10:38:06 restore Initial Maven and .gitignore configuration #1 from checkpoint N: the committer re-commits all five producers successfully.
• 10:38:22 writers 0-3 get the typed OffsetAlreadyExists on their first append (appends had advanced past the checkpointed offset before the crash), log "cannot use stream ... Discarding this stream", finalize their streams and create new ones. This is the designed path and it worked.
• 10:39:55 checkpoint N+1 is snapshotted but never completes: the writers fail again at 10:43 and 10:45 (MaximumRequestCallbackWaitTimeExceededException, the first append after the snapshot waited 5 minutes in the client's inflight queue).
• From 10:47:51, every ~30 s for 14 hours: "Restoring state" -> "Committing records appended by producer 0..4" -> Commit operation failed, caused by InvalidArgumentException: ... Stream has been finalized and cannot be appended: <table>::<uuid>:s14 (and :s0, :s4, :s12): exactly the four streams finalized at 10:38:22. The writer never ran again during the loop ("Snapshotting state": 0 occurrences; "Discarding this stream": 4 in total). Recovery required discarding the checkpoints and starting fresh.

Note that the "Stream has been finalized and cannot be appended" message is the same for a rejected flush and a rejected append, which makes the log look like a writer problem. An append to a finalized stream actually surfaces as the typed StreamFinalizedException with the bundled client (3.2.0, same in 1.1.0, 1.2.0 and main), which the writer already recognizes.

Fix (PR follows):

• Writer: never finalize a stream that checkpoint state references. discardStream() only abandons the stream; close() finalizes only a stream created after the last snapshot (no checkpoint references it). Uncommitted appends on a checkpointed stream are never flushed and are replaced by the checkpoint replay, as today.
• Committer: remember the stream last committed for each producer and finalize the previous stream, best effort, after a committable for a different stream is committed. A commit runs only after its checkpoint completed, so no restorable checkpoint references the previous stream at that point. A stream the committer never sees replaced (no committable for that producer in that checkpoint, or an aborted checkpoint) stays open until BigQuery's idle TTL; this only concerns the server-side stream object on failure paths, the client-side StreamWriter is closed as before.

This also closes a data-loss path: the first append after a checkpoint takes the same restored-stream path, so a timeout there (handled since #337 by discarding the stream) could finalize the checkpointed stream before the committer's notifyCheckpointComplete flush ran.

What the fix does not cover: a deliberate restore from a checkpoint older than the latest completed one, or from a savepoint taken before the producer moved to a new stream, can still name a stream that has since been finalized (or whose flush cursor has since advanced, which is rejected with ALREADY_EXISTS). Those rollbacks already failed before the change; the committer would need an "already flushed" tolerance to support them, which is a separate discussion.

[seokjin0414]

PR with the fix for both modules: #347