apache/flink · error · IllegalStateException

Writer has already been opened as {writeType} type, but tryi

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.

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)

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" operatio
• can not recover from the pendingFileRecoverable
• Could not create writer state serializer.
• Could not create committable serializer.
• Unrecognized version or corrupt state: {version}

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