{"record":{"id":"6842bd9e3d38ba58","repo":"apache/beam","slug":"stream-has-been-finished-can-not-add-any-more-elements","errorCode":null,"errorMessage":"Stream has been finished. Can not add any more elements.","messagePattern":"Stream has been finished\\. Can not add any more elements\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/util/BufferedElementCountingOutputStream.java","lineNumber":134,"sourceCode":"    if (finished) {\n      return;\n    }\n    flush();\n    // Finish the stream with the terminatorValue.\n    VarInt.encode(terminatorValue, os);\n    if (!BUFFER_POOL.offer(buffer)) {\n      // The pool is full, we can't store the buffer. We just drop the buffer.\n    }\n    finished = true;\n  }\n\n  /**\n   * Marks that a new element is being output. This allows this output stream to use the buffer if\n   * it had previously overflowed marking the start of a new block of elements.\n   */\n  public void markElementStart() throws IOException {\n    if (finished) {\n      throw new IOException(\"Stream has been finished. Can not add any more elements.\");\n    }\n    count++;\n  }\n\n  @Override\n  public void write(int b) throws IOException {\n    if (finished) {\n      throw new IOException(\"Stream has been finished. Can not write any more data.\");\n    }\n    if (count == 0) {\n      os.write(b);\n      return;\n    }\n\n    if (buffer.hasRemaining()) {\n      buffer.put((byte) b);\n    } else {\n      outputBuffer();","sourceCodeStart":116,"sourceCodeEnd":152,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BufferedElementCountingOutputStream.java#L116-L152","documentation":"BufferedElementCountingOutputStream counts elements written and emits length markers; once finished() is called the stream is closed for further output. markElementStart() checks the finished flag and throws IOException if a new element is started after finish. The stream is a one-shot sink: finishing is terminal.","triggerScenarios":"Calling markElementStart() after finish() — e.g. a caller keeps encoding elements after closing, or reuses the same output-stream wrapper for a second batch without creating a new instance.","commonSituations":"Reuse of a stream across batches or retries after finish; encode() loops whose termination condition runs one iteration past finish; custom sinks built on this stream that flush-then-finish but still receive data.","solutions":["Create a new BufferedElementCountingOutputStream for each batch instead of reusing a finished instance.","Ensure finish() is called only after all elements have been marked and written; move finish() to the true end of the encode loop.","If reuse is required, track a 'finished' state in your wrapper and reinitialize the underlying stream before further writes."],"exampleFix":"// before\nstream.markElementStart(); writeElement(stream);\nstream.finish();\nstream.markElementStart(); // IOException\n// after\nstream.markElementStart(); writeElement(stream);\nstream.finish();\nstream = new BufferedElementCountingOutputStream(os, 20);\nstream.markElementStart(); writeElement(stream);","handlingStrategy":"validation","validationCode":"if (!stream.finished /* or your own flag */) { stream.markElementStart(); }","typeGuard":null,"tryCatchPattern":"try (BufferedElementCountingOutputStream s = ...) {\n  // write all elements\n} // finish via close; never write afterwards","preventionTips":["Never reuse a finished stream; create a new one per batch.","Call finish() only after the last element is fully written.","Wrap lifecycle in try-with-resources or a owner object that serializes writes."],"tags":["java","apache-beam","io","stream"],"backgroundTag":"invalid-state-transition","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"}