fix dataloss bug in batch Storage API sink.

[reuvenlax]

Two issues fixed:

1. Retry behavior was incorrect. We created a new stream abandoning the old, but then only retried the current append.
2. Offset was inadvertently reset to 0 the first time we called append, which meant that the second append would always fail with invalid offset. This triggered the data-loss bug in 1.

We fixed the offset bug and no longer create a new stream on failure. In cases where the stream is known to be unusable (e.g. it's been deleted), we raise an exception which forces retry of the entire work item.

[reuvenlax]

R: @johnjcasey

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

[johnjcasey]

LGTM. This will break storage api writes for now, but we will no longer have false positive results

[johnjcasey]

LGTM

[ahmedabu98]

Would it be helpful for future debugging if we include the error's stacktrace here?

[johnjcasey]

fixes #26521 and #26520

[kennknowles]

Why isn't there a test for this?

[reuvenlax]

Good question - there should be a unit test, and I believe this would be fairly easy to reproduce in a unit test.

…

On Wed, May 3, 2023 at 1:20 PM Kenn Knowles ***@***.***> wrote: Why isn't there a test for this? — Reply to this email directly, view it on GitHub <#26503 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFAYJVKPIYSAZ67MOZQOK5DXEK42HANCNFSM6AAAAAAXTISF4A> . You are receiving this because you modified the open/close state.Message ID: ***@***.***>

[ahmedabu98]

We only have one test case that indirectly tests for multiple stream appends, and it uses streaming mode (while this error occurs in batch):

beam/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java

Lines 1971 to 1976 in 03ecb15

	public void updateTableSchemaTest(boolean useSet) throws Exception {
	assumeTrue(useStreaming);
	assumeTrue(useStorageApi);
	// Make sure that GroupIntoBatches does not buffer data.
	p.getOptions().as(BigQueryOptions.class).setStorageApiAppendThresholdBytes(1);

[kennknowles]

Yea let's add a batch test that reproduces the issue at the commit before the fix, and confirms it fixed. Keep the bug open until the test is in.

[ahmedabu98]

I'll write up the test and tag @reuvenlax to review