pkuzmickas opened a new pull request, #41290: URL: https://github.com/apache/beam/pull/41290
When a task is cancelled while `BigtableWriterFn` is in `finishBundle`, the interrupt reaches `BigtableWriterImpl.close()` while it waits for the batcher to close. `close()` throws, but leaves the writer open (`bulkMutation` set) and its `stopwatch` running. `tearDown()` then calls `close()` again, which fails with `IllegalStateException: This stopwatch is already running`, and `tearDown()` rethrows it. On the Flink runner, an exception while cleaning up a cancelled task is fatal: Flink shuts down the whole TaskManager, so every rescale, failover or stop of a job writing with `BigtableIO` can take TaskManagers down with it. We saw this on Flink 2.3 with Beam 2.72: TaskManagers exited on cancellation with `FATAL` and this stack trace. Changes: - `BigtableWriterImpl.close()` uses a new `Stopwatch` per call, so closing again after a failed or interrupted close works. - `BigtableWriterFn.tearDown()` logs a failure to close the writer instead of rethrowing it. The writer is only still open in teardown if the bundle failed or was cancelled, so its mutations are retried with the bundle; teardown is best effort anyway. `serviceEntry` is still closed in `finally`. Testing: `BigtableServiceImplTest.testCloseAgainAfterInterruptedClose` closes a writer on an interrupted thread, then closes it again. Without the fix the second close throws `IllegalStateException: This stopwatch is already running.`; with it, `BigtableServiceImplTest` passes (14 tests). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
