prestodb/presto · error · UnsupportedOperationException

Table creation is not supported by the kafka connector

Error message

Table creation is not supported by the kafka connector

What it means

The kafka connector does not implement table creation, so KafkaPageSinkProvider.createPageSink refuses to serve a CREATE TABLE AS INSERT commit; this is a capability guard, not a runtime fault — the target table must already exist as a Kafka topic mapped through the catalog.

Source

Thrown at presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaPageSinkProvider.java:58

public class KafkaPageSinkProvider
        implements ConnectorPageSinkProvider
{
    private final DispatchingRowEncoderFactory encoderFactory;
    private final KafkaProducerFactory producerFactory;
    private final KafkaClusterMetadataSupplier kafkaClusterMetadataSupplier;

    @Inject
    public KafkaPageSinkProvider(DispatchingRowEncoderFactory encoderFactory, KafkaProducerFactory producerFactory, KafkaClusterMetadataSupplier kafkaClusterMetadataSupplier)
    {
        this.encoderFactory = requireNonNull(encoderFactory, "encoderFactory is null");
        this.producerFactory = requireNonNull(producerFactory, "producerFactory is null");
        this.kafkaClusterMetadataSupplier = requireNonNull(kafkaClusterMetadataSupplier, "kafkaClusterMetadataSupplier is null");
    }

    @Override
    public ConnectorPageSink createPageSink(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorOutputTableHandle outputTableHandle, PageSinkContext pageSinkContext)
    {
        throw new UnsupportedOperationException("Table creation is not supported by the kafka connector");
    }

    @Override
    public ConnectorPageSink createPageSink(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorInsertTableHandle insertTableHandle, PageSinkContext pageSinkContext)
    {
        requireNonNull(insertTableHandle, "tableHandle is null");
        KafkaTableHandle handle = (KafkaTableHandle) insertTableHandle;

        ImmutableList.Builder<EncoderColumnHandle> keyColumns = ImmutableList.builder();
        ImmutableList.Builder<EncoderColumnHandle> messageColumns = ImmutableList.builder();
        handle.getColumns().forEach(col -> {
            if (col.isInternal()) {
                throw new IllegalArgumentException(format("unexpected internal column '%s'", col.getName()));
            }
            if (col.isKeyCodec()) {
                keyColumns.add(col);
            }
            else {

View on GitHub (pinned to 55bb57d202)

Solutions

1. Pre-create the Kafka topic and register the table in the kafka connector's topic definition file
2. Use a connector that supports DDL if you need CREATE TABLE AS

When it happens

Trigger: Thrown at presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaPageSinkProvider.java:58 when the library encounters an invalid state.

Common situations: See trigger scenarios.

AI-assisted analysis of prestodb/presto@55bb57d202 (2026-09-04). Data as JSON: /api/errors/24e631d94f31cd34. Report an issue: GitHub.