{"record":{"id":"24e631d94f31cd34","repo":"prestodb/presto","slug":"table-creation-is-not-supported-by-the-kafka-conne","errorCode":null,"errorMessage":"Table creation is not supported by the kafka connector","messagePattern":"Table creation is not supported by the kafka connector","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaPageSinkProvider.java","lineNumber":58,"sourceCode":"public class KafkaPageSinkProvider\n        implements ConnectorPageSinkProvider\n{\n    private final DispatchingRowEncoderFactory encoderFactory;\n    private final KafkaProducerFactory producerFactory;\n    private final KafkaClusterMetadataSupplier kafkaClusterMetadataSupplier;\n\n    @Inject\n    public KafkaPageSinkProvider(DispatchingRowEncoderFactory encoderFactory, KafkaProducerFactory producerFactory, KafkaClusterMetadataSupplier kafkaClusterMetadataSupplier)\n    {\n        this.encoderFactory = requireNonNull(encoderFactory, \"encoderFactory is null\");\n        this.producerFactory = requireNonNull(producerFactory, \"producerFactory is null\");\n        this.kafkaClusterMetadataSupplier = requireNonNull(kafkaClusterMetadataSupplier, \"kafkaClusterMetadataSupplier is null\");\n    }\n\n    @Override\n    public ConnectorPageSink createPageSink(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorOutputTableHandle outputTableHandle, PageSinkContext pageSinkContext)\n    {\n        throw new UnsupportedOperationException(\"Table creation is not supported by the kafka connector\");\n    }\n\n    @Override\n    public ConnectorPageSink createPageSink(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorInsertTableHandle insertTableHandle, PageSinkContext pageSinkContext)\n    {\n        requireNonNull(insertTableHandle, \"tableHandle is null\");\n        KafkaTableHandle handle = (KafkaTableHandle) insertTableHandle;\n\n        ImmutableList.Builder<EncoderColumnHandle> keyColumns = ImmutableList.builder();\n        ImmutableList.Builder<EncoderColumnHandle> messageColumns = ImmutableList.builder();\n        handle.getColumns().forEach(col -> {\n            if (col.isInternal()) {\n                throw new IllegalArgumentException(format(\"unexpected internal column '%s'\", col.getName()));\n            }\n            if (col.isKeyCodec()) {\n                keyColumns.add(col);\n            }\n            else {","sourceCodeStart":40,"sourceCodeEnd":76,"githubUrl":"https://github.com/prestodb/presto/blob/55bb57d202de3b926896fa966c2c4a44c779634e/presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaPageSinkProvider.java#L40-L76","documentation":"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.","triggerScenarios":"Thrown at presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaPageSinkProvider.java:58 when the library encounters an invalid state.","commonSituations":"See trigger scenarios.","solutions":["Pre-create the Kafka topic and register the table in the kafka connector's topic definition file","Use a connector that supports DDL if you need CREATE TABLE AS"],"exampleFix":null,"handlingStrategy":"validation","validationCode":null,"typeGuard":null,"tryCatchPattern":null,"preventionTips":[],"tags":[],"backgroundTag":null,"analyzedSha":"55bb57d202de3b926896fa966c2c4a44c779634e","analyzedAt":"2026-09-04T12:50:26.162Z","contentChangedAt":"2026-09-04T12:50:26.162Z","schemaVersion":2},"datasetVersion":"2026-09-11T21:17:09.523Z"}