presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaMetadata.java/KafkaMetadata/getTableHandle
class KafkaMetadata:    @Override
    public KafkaTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName)
    {
        KafkaTopicDescription table = tableDescriptions.get(schemaTableName);
        if (table == null) {
            return null;
        }

        return new KafkaTableHandle(
                schemaTableName.getSchemaName(),
                schemaTableName.getTableName(),
                table.getTopicName(),
                getDataFormat(table.getKey()),
                getDataFormat(table.getMessage()),
                table.getKey().flatMap(KafkaTopicFieldGroup::getDataSchema),
                table.getMessage().flatMap(KafkaTopicFieldGroup::getDataSchema));
    }