{"record":{"id":"ec280d54022987af","repo":"apache/beam","slug":"bulkmutation-took-too-long-to-close","errorCode":null,"errorMessage":"BulkMutation took too long to close","messagePattern":"BulkMutation took too long to close","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableServiceImpl.java","lineNumber":567,"sourceCode":"          // set a timeout waiting for the future.\n          ApiFuture<Void> future = bulkMutation.closeAsync();\n          if (Duration.ZERO.isShorterThan(closeWaitTimeout)) {\n            future.get(closeWaitTimeout.getMillis(), TimeUnit.MILLISECONDS);\n          } else {\n            future.get();\n          }\n          bulkSize.update(outstandingMutations);\n          outstandingMutations = 0;\n          stopwatch.stop();\n          latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));\n        } catch (BatchingException e) {\n          // Ignore batching failures because element failures are tracked as is in\n          // BigtableIOWriteFn.\n          // TODO: Bigtable client already tracks BatchingExceptions, use BatchingExceptions\n          // instead of tracking them separately in BigtableIOWriteFn.\n        } catch (TimeoutException e) {\n          // We fail because future.get() timed out\n          throw new IOException(\"BulkMutation took too long to close\", e);\n        } catch (ExecutionException e) {\n          throw new IOException(\"Failed to close batch\", e.getCause());\n        } catch (InterruptedException e) {\n          Thread.currentThread().interrupt();\n          // We fail since close() operation was interrupted.\n          throw new IOException(e);\n        }\n        bulkMutation = null;\n      }\n    }\n\n    @Override\n    public CompletableFuture<MutateRowResponse> writeRecord(\n        KV<ByteString, Iterable<Mutation>> record) throws IOException {\n\n      com.google.cloud.bigtable.data.v2.models.Mutation mutation =\n          com.google.cloud.bigtable.data.v2.models.Mutation.fromProtoUnsafe(record.getValue());\n","sourceCodeStart":549,"sourceCodeEnd":585,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableServiceImpl.java#L549-L585","documentation":"BigtableServiceImpl.close flushes and waits for the in-flight BulkMutation batcher to complete. If the future returned by the batcher's close/flush does not finish before the configured timeout, a TimeoutException is wrapped into an IOException with this message, failing the write.","triggerScenarios":"Closing the Bigtable writer while bulk mutations are still being flushed; the batcher's future.get() times out because Cloud Bigtable is slow, overloaded, or the network is degraded, or because the flush timeout is too short for the volume of pending mutations.","commonSituations":"Large bulkloads writing millions of rows where the final flush takes longer than the timeout; Bigtable cluster throttling; narrow per-shutdown timeout budgets in streaming pipelines.","solutions":["Increase the Bigtable write flush/close timeout (e.g. via BigtableIO write flow control / max outstanding elements options)","Reduce write throughput or batch size so flushes complete within the timeout","Check Cloud Bigtable instance CPU utilization and scale the cluster if throttling","Retry the affected bundle; partial failures are tracked per-element in BigtableIOWriteFn"],"exampleFix":"// before\npipeline.apply(BigtableIO.write().withBigtableOptions(opts)); // default timeouts\n// after\nopts = opts.toBuilder()\n    .setBulkOptions(BulkOptions.newBuilder().setMaxElementCountPerRow(100).build())\n    .build();\npipeline.apply(BigtableIO.write().withBigtableOptions(opts));","handlingStrategy":"try-catch","validationCode":"// size the flush budget against pending work\nlong pending = batcher.getOutstandingElementCount();\nif (pending > threshold) { /* increase timeout or throttle writes */ }","typeGuard":null,"tryCatchPattern":"try {\n  writer.close();\n} catch (IOException e) {\n  if (e.getCause() instanceof TimeoutException) {\n    // retry flush or extend timeout\n  }\n}","preventionTips":["Configure flow control (withMaxOutstandingElementCount) to bound pending mutations","Monitor Bigtable instance CPU/latency and scale before bulkloads","Keep close-timeout proportional to volume of pending writes"],"tags":["gcp","bigtable","timeout","io"],"backgroundTag":"request-timeout","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}