{"record":{"id":"f132285188c9542b","repo":"apache/beam","slug":"failed-to-close-batch","errorCode":null,"errorMessage":"Failed to close batch","messagePattern":"Failed to close batch","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":569,"sourceCode":"          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\n      RowMutationEntry entry = RowMutationEntry.createFromMutationUnsafe(record.getKey(), mutation);\n","sourceCodeStart":551,"sourceCodeEnd":587,"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#L551-L587","documentation":"BigtableServiceImpl.close waits for the BulkMutation batcher's completion future. If that future completes exceptionally, the ExecutionException's cause is wrapped into an IOException with this message, meaning the final flush of pending mutations failed.","triggerScenarios":"Calling close() on the Bigtable writer when the batcher's flush future fails — typically due to gRPC errors from Cloud Bigtable during the final flush (e.g. server errors, dead peer, quota limits).","commonSituations":"Network drop or Bigtable outage at pipeline shutdown; invalid table name not caught earlier; mutation payload rejected at flush time.","solutions":["Inspect e.getCause() on the resulting IOException to find the underlying gRPC status","Re-run the pipeline — element failures are tracked in BigtableIOWriteFn and written to the failure output","Validate tableId/instanceId/projectId are consistent across the pipeline","Ensure the table and column families exist before writing"],"exampleFix":"// before\ntable.exists(); // assume table exists, let flush fail\n// after\nif (!service.checkTableExists(tableId)) {\n  createTableWithFamilies(projectId, instanceId, tableId, families);\n}","handlingStrategy":"retry","validationCode":"// pre-flight: confirm table exists before writing\nif (!factory.tableExists(tableId)) { createTable(projectId, instanceId, tableId, families); }","typeGuard":null,"tryCatchPattern":"try {\n  writer.close();\n} catch (IOException e) {\n  Throwable cause = e.getCause(); // ExecutionException cause from flush future\n  if (cause instanceof ApiException) { /* inspect gRPC status, retry */ }\n}","preventionTips":["Ensure table and column families exist before writing","Rely on BigtableIOWriteFn per-element failure tracking and retry the failing bundles","Watch network stability between the runner and Cloud Bigtable"],"tags":["gcp","bigtable","io","flush"],"backgroundTag":"database-write-failed","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"}