R2DBC Postgres SQL Enum Issue

As part of learning r2DBC i have come across an issue facing with Enum conversion. I am using PostgreSQL here. When reading data for Film where rating is PG-13 and NC-17( anything with dash ) i am facing issues.

Below is my schema for table

create table film
(
film_id          integer       default nextval('film_film_id_seq'::regclass) not null
    constraint film_pkey
        primary key,
title            varchar(255)                                                not null,
description      text,
release_year     year,
language_id      smallint                                                    not null
    constraint film_language_id_fkey
        references language
        on update cascade on delete restrict,
rental_duration  smallint      default 3                                     not null,
rental_rate      numeric(4, 2) default 4.99                                  not null,
length           smallint,
replacement_cost numeric(5, 2) default 19.99                                 not null,
rating           mpaa_rating   default 'G'::mpaa_rating,
last_update      timestamp     default now()                                 not null,
special_features text[]
);

And the mpaa_rating is defined as

create type mpaa_rating as enum ('G', 'PG', 'PG-13', 'R', 'NC-17');

This is my code which registers the converters in my Configuration

@Configuration
@EnableTransactionManagement
@EnableR2dbcRepositories
@EnableR2dbcAuditing
public class DVDRentalDBConfiguration extends AbstractR2dbcConfiguration {

@Bean
public ConnectionFactory connectionFactory() {
    System.out.println("Initializing postgreSQL connection factory");
    return new PostgresqlConnectionFactory(
            PostgresqlConnectionConfiguration.builder()
                    .host("localhost")
                    .database("dvdrental")
                    .username("postgres")
                    .password("postgres")
                    .codecRegistrar(EnumCodec.builder().withEnum("mpaa_rating", Rating.class).build())
                    .build()
    );
}

@Override
protected List<Object> getCustomConverters() {
    return Collections.singletonList(new RatingWritingConverter());
}

@Bean
ReactiveTransactionManager transactionManager(ConnectionFactory connectionFactory) {
    System.out.println("Initializing postgreSQL connection factory");
    return new R2dbcTransactionManager(connectionFactory);
}
}

My code for retrieving is pretty simple

 private Mono<FilmModel> getFilmById(Long id) {
    return filmRepository.findById(id).switchIfEmpty(Mono.error(DataFormatException::new));
}
  

Adding the exception which is thrown https://gist.github.com/harryalto/bd51bbcdd081868c5064c808d08205e4

I tried researching stack overflow but couldn't figure out the issue. Any help is greatly appreciated.

If you are using Spring Boot/Spring Data R2dbc to map table to POJO, you can skip the enum definition in Postgres, by default Spring Data R2dbc will handle the enum as varchar/char in db side, and use Enum in java POJO, check my example, and schema sql script and mapped entity class. Spring Boot registered mapping converter to convert them automatically.

If you would like to handle the Enum type yourself, check this example.