apache/flink · error · IllegalStateException

Writer has already been opened as

Error message

Writer has already been opened as {writeType} type, but trying to reopen it as {type} type.

What it means

Thrown by OutputStreamBasedPartFileWriter.ensureWriteType when the writer has already committed to one write mode (RECORD_WISE via write() or OUTPUT_STREAM via asOutputStream()) and a call attempts to switch to the other. A single part-file writer supports only one mode for its lifetime because the underlying BulkWriter or OutputStream contract assumes exclusive access.

Solutions

1. Pick one write mode per writer instance and stick with it — do not interleave write() and asOutputStream().
2. If both modes are needed, open a new in-progress file for the second mode rather than reusing the same writer.
3. Audit custom compaction or BucketWriter implementations to ensure they don't mix the two APIs.

Example fix

// before — mixing write types on the same writer
writer.write(record, now);
OutputStream os = writer.asOutputStream(); // throws IllegalStateException
// after — use one mode per writer
writer.write(record, now);
// or open a dedicated stream-based writer
OutputStreamBasedPartFileWriter<?, ?> streamWriter =
    (OutputStreamBasedPartFileWriter<?, ?>) bucketWriter.openNewInProgressFile(bucketId, path, ts);
OutputStream os = streamWriter.asOutputStream();

// Track which write mode the writer is in before switching
// Do not interleave write() and asOutputStream() on the same writer
if (writerHasBeenUsedForRecords) {
    throw new IllegalStateException("Cannot obtain OutputStream after record-wise writes");
}

When it happens

Trigger: Calling write(element, time) (which sets RECORD_WISE) followed by asOutputStream() (which sets OUTPUT_STREAM) on the same InProgressFileWriter instance, or vice versa.

Common situations: A custom BucketWriter or compaction logic that tries to write elements via the record API and also obtain the raw OutputStream on the same file; mixing BulkWriter.addElement with direct stream writes in the same bucket lifecycle.

Related errors

• Bulk Part Writers do not support "pause and resume"…
• can not recover from the pendingFileRecoverable
• At least one trigger condition must be configured.
• Cannot clean commit: File has trailing junk data.
• Cannot clean commit: Staging file does not exist.

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/48d378d50d7d9739. Report an issue: GitHub.

Appendix: source

Thrown at flink-connectors/flink-file-sink-common/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/OutputStreamBasedPartFileWriter.java:101

    }

    @Override
    public long getSize() throws IOException {
        return currentPartStream.getPos();
    }

    @Override
    public OutputStream asOutputStream() throws IOException {
        ensureWriteType(Type.OUTPUT_STREAM);
        return currentPartStream;
    }

    protected void ensureWriteType(Type type) {
        if (type != this.writeType) {
            if (this.writeType == null) {
                this.writeType = type;
            } else {
                throw new IllegalStateException(
                        "Writer has already been opened as "
                                + writeType
                                + " type, but trying to reopen it as "
                                + type
                                + " type.");
            }
        }
    }

    abstract static class OutputStreamBasedBucketWriter<IN, BucketID>
            implements BucketWriter<IN, BucketID> {

        private final RecoverableWriter recoverableWriter;

        OutputStreamBasedBucketWriter(final RecoverableWriter recoverableWriter) {
            this.recoverableWriter = recoverableWriter;
        }

View on GitHub (pinned to 2f3c205e92)