Why I Reworked a BigQuery Sink from Pending Streams to Buffered Streams

While working on the BigQuery Sink Connector for Apache SeaTunnel, I initially thought the main challenge would be straightforward: write rows into BigQuery correctly.

But as the implementation evolved, I realized the real problem was deeper than just calling the BigQuery API. The hard part was aligning BigQuery’s external write visibility model with SeaTunnel’s checkpoint and restore lifecycle.

This post summarizes the design issue I encountered while using the BigQuery Storage Write API, why the initial pending-stream design was not strong enough for checkpoint recovery, and why I eventually redesigned the batch write path around buffered streams.

Initial Design: Pending Streams for Batch Writes

BigQuery Storage Write API provides several write stream types. At first, Pending Stream looked like a natural fit for batch writes.

The flow is roughly:

Create PENDING stream
AppendRows
FinalizeWriteStream
BatchCommitWriteStreams

Data written to a pending stream is not visible to readers until BatchCommitWriteStreams is called. Since BigQuery can commit multiple pending streams atomically, this looked like a good match for a checkpoint-based sink commit protocol.

The initial implementation looked roughly like this:

@Override
public Optional<BigQueryCommitInfo> prepareCommit() {
    flush();
    streamWriter.finalizeStream();
    return Optional.of(new BigQueryCommitInfo(streamWriter.getStreamName()));
}
@Override
public List<Void> snapshotState(long checkpointId) {
    this.streamWriter.close();
    this.streamWriter = BigQueryBatchWriter.of(client, config);
    return Collections.emptyList();
}

At first glance, this looked similar to a two-phase commit protocol.

prepareCommit()  -> finalize stream
commit()         -> BatchCommitWriteStreams

However, during review, a subtle but important problem surfaced.

The Problem: Ownership Gap After a Failed Checkpoint

In SeaTunnel’s sink lifecycle, prepareCommit() is called before the checkpoint is fully completed. The issue is that prepareCommit() already performs an external side effect by calling FinalizeWriteStream.

Consider this scenario:

1. Checkpoint N starts.
2. prepareCommit() finalizes stream A.
3. commitInfo(stream A) is created.
4. The job fails before checkpoint N completes.
5. The job restores from the last 
completed checkpoint N-1.

After restore, restoreWriter(states)receives only the state of the last completed checkpoint. Failed checkpoint state is discarded.

This means stream A, which was finalized during the failed checkpoint N, is not part of the restored state.

That is the core issue.

FinalizeWriteStream has already been called,
but the checkpoint does not contain a durable decision
about whether this finalized stream should be committed,
abandoned, or cleaned up.

At first, I thought that since pending-stream data is invisible before BatchCommitWriteStreams, abandoning the stream might be acceptable. However, that explanation was not strong enough. To claim clear checkpoint recovery semantics, the connector needs a more explicit story about external side effects.

Why restoreWriter() Should Not Commit Failed-Checkpoint Streams

One seemingly simple solution is to discover and commit the finalized pending stream during restore.

But this is unsafe.

Remember that restore starts from the last completed checkpoint, N-1. Therefore, records that were processed during checkpoint N may be replayed after restore.

If restoreWriter() discovers stream A and commits it, the following can happen:

records in stream A become visible in BigQuery
+
the same records are replayed after restore from checkpoint N-1
+
the replayed records are written again and committed later

This could produce duplicate rows if the connector commits a stream that belongs to a failed checkpoint.

So the important point is:

restoreWriter() receives only successful checkpoint state.
A stream finalized by a failed checkpoint is not part of that state.
If we try to commit such an external stream during restore,
we may publish records that the engine is about to replay.

Therefore, streams finalized by failed checkpoints should not be blindly committed during restore.

The issue is not that Pending Streams always cause duplicates; the issue is that the connector has no durable decision for finalized-but-uncommitted streams created by failed checkpoints.

Redesign: Switching to Buffered Streams

To avoid this ownership gap, I redesigned the batch write path to use Buffered Streams instead of Pending Streams.

Buffered Streams do not commit the whole stream through a finalize-and-batch-commit protocol. Instead, they make data visible up to a specific stream offset.

The flow looks roughly like this:

Create BUFFERED stream
AppendRows(offset=N)
FlushRows(offset=M)

Data appended to a buffered stream is not immediately visible. It becomes visible only after FlushRows advances visibility up to a specific offset.

This model maps much better to checkpoint-based recovery.

A checkpoint represents the processing position of the engine. A buffered stream offset represents the external write position in BigQuery.

The new design looks like this:

writer state:
  streamName
  nextOffset
  checkpointId
prepareCommit(checkpointId):
  flush()
  return BigQueryCommitInfo(streamName, flushOffset = nextOffset - 1)
snapshotState(checkpointId):
  return BigQuerySinkState(streamName, nextOffset, checkpointId)
commit(commitInfo):
  FlushRows(streamName, flushOffset)
restoreWriter(states):
  select the latest completed checkpoint state
  restore writer with streamName + nextOffset

Now the external write position is represented by:

streamName + nextOffset

This position is stored in checkpoint state. After restore, the writer can resume from the last completed checkpoint’s external write position.

Understanding State and CommitInfo

One important lesson was that writer state and commit info are related, but they mean different things.

Write on Medium
The writer state represents where the writer should resume appending after restore.

BigQuerySinkState {
    String streamName;
    long nextOffset;
    long checkpointId;
}

The commit info represents what should become visible in BigQuery after the checkpoint completes.

BigQueryCommitInfo {
    String streamName;
    long flushOffset;
}

For example:

writer state:
  streamName = S
  nextOffset = 100
commit info:
  streamName = S
  flushOffset = 99

This means:

The writer has appended rows up to offset 99.
If the checkpoint completes, BigQuery should flush visibility up to offset 99.
If the job restores from this checkpoint, the writer should resume from offset 100.

This separation made the recovery model much clearer.

Managing Offsets Correctly

The most important detail in the buffered-stream design is offset management.

The offset is not a record identifier. It is not a primary key. It is also not a checkpoint id.

The offset is the append position inside a specific BigQuery write stream.

If there are multiple parallel writers, each writer should own its own stream, and each stream should manage its offset independently.

For example:

writer-0:
  stream S0
  nextOffset = 100
writer-1:
  stream S1
  nextOffset = 250

Even if the checkpoint id is 10, the BigQuery append offset should not be 10. The checkpoint id is only metadata. The BigQuery offset must represent the append position inside that stream.

So the model is:

checkpointId:
  used to identify and select checkpoint state
nextOffset:
  used as the next append position in the BigQuery stream
flushOffset:
  nextOffset - 1, used by FlushRows

This distinction is important, especially when the sink runs with parallelism.

Committer Uses FlushRows Instead of BatchCommitWriteStreams

With Pending Streams, the committer used BatchCommitWriteStreams.

With Buffered Streams, the committer uses FlushRows.

FlushRowsRequest request =
        FlushRowsRequest.newBuilder()
                .setWriteStream(info.getStreamName())
                .setOffset(Int64Value.of(info.getFlushOffset()))
                .build();
FlushRowsResponse response = client.flushRows(request);

This makes rows visible up to the requested offset.

So after a checkpoint completes, the committer advances BigQuery visibility up to the checkpoint’s flushOffset.

Keeping CDC Separate

This redesign only applies to the batch write path.

I did not change the CDC path to use Buffered Streams. BigQuery CDC ingestion has its own semantics around _CHANGE_TYPE, _CHANGE_SEQUENCE_NUMBER, and primary keys. It should not be forced into the same checkpoint-offset model as batch writes.

The resulting structure is:

batch mode:
  Buffered Stream
  streamName + nextOffset in checkpoint state
  FlushRows on commit
cdc / streaming mode:
  existing streaming path
  BigQuery CDC semantics

This keeps the batch recovery model explicit without changing CDC behavior unnecessarily.

Testing Challenges

This change depends on real BigQuery Storage Write API behavior, especially Buffered Streams, explicit offsets, and FlushRows. A local emulator was not sufficient to validate these semantics reliably.

I tested the updated batch path against a real BigQuery environment.

The verified scenarios included:

- creating a Buffered Stream
- appending rows with explicit offsets
- running a batch write with checkpoint enabled
- verifying that rows become visible in BigQuery after FlushRows

A fully deterministic failure-recovery E2E test is much harder. It would require controlling checkpoint barrier timing, injecting failure at a precise point, restoring the job, and verifying BigQuery visibility and duplicates.

For automated tests, a more practical approach is to cover the recoverable metadata path:

- BigQuerySinkState serialization
- selecting the latest state by checkpointId
- advancing nextOffset only after append success
- creating FlushRows commit info

This does not replace a full failure-recovery E2E test, but it validates the most important internal recovery metadata.

Lessons Learned

The biggest lesson I learned is that connector exactly-once semantics are not just about calling a commit API.

A connector must answer questions like:

1. When does data become visible in the external system?
2. What happens to external side effects if a checkpoint fails?
3. What external write position should the writer restore from?
4. What does writer state mean?
5. What does commitInfo mean?
6. How should offsets behave when append fails and retries happen?

The initial Pending Stream design looked natural from BigQuery’s batch commit perspective. But when combined with SeaTunnel’s checkpoint and restore lifecycle, it created an ownership gap for finalized streams after failed checkpoints.

The Buffered Stream design is more complex, but it gives the connector a clearer recovery model by storing the external write position as streamName + offset.

Closing Thoughts

This work was not just about adding a BigQuery Sink Connector.

It was about aligning an external system’s write visibility model with a stream processing engine’s checkpoint lifecycle.

The first design was not perfect. Review revealed an important failure scenario. But by understanding the issue, revisiting BigQuery stream types, and redesigning the batch path around Buffered Streams, the connector’s recovery semantics became much clearer.

Open source review can be painful, but it often forces us to think beyond whether the code works in the happy path. It pushes us to reason about boundaries, failure windows, and system guarantees.

For me, this BigQuery Sink work was exactly that kind of experience.