Mapping to JSON Fields with Spring Data R2DBC and Reactive MySQL Driver

Question

I'm trying to use JSON data type to store json data in a specific column in a MySQL database.

I followed this guide, and got storing and reading JSON working fine but now I get this error when trying to save another entity that have a boolean column/field:

Caused by: org.springframework.core.convert.ConverterNotFoundException: No converter found capable of converting from type [java.lang.Byte] to type [boolean]
    at org.springframework.core.convert.support.GenericConversionService.handleConverterNotFound(GenericConversionService.java:322)
    at org.springframework.core.convert.support.GenericConversionService.convert(GenericConversionService.java:195)
    at org.springframework.core.convert.support.GenericConversionService.convert(GenericConversionService.java:175)
    at org.springframework.data.r2dbc.convert.MappingR2dbcConverter.getPotentiallyConvertedSimpleRead(MappingR2dbcConverter.java:277)
    at org.springframework.data.r2dbc.convert.MappingR2dbcConverter.readValue(MappingR2dbcConverter.java:201)
    at org.springframework.data.r2dbc.convert.MappingR2dbcConverter.readFrom(MappingR2dbcConverter.java:182)
    at org.springframework.data.r2dbc.convert.MappingR2dbcConverter.read(MappingR2dbcConverter.java:136)
    at org.springframework.data.r2dbc.convert.MappingR2dbcConverter.read(MappingR2dbcConverter.java:119)
    at org.springframework.data.r2dbc.convert.EntityRowMapper.apply(EntityRowMapper.java:46)
    at org.springframework.data.r2dbc.convert.EntityRowMapper.apply(EntityRowMapper.java:29)
    at dev.miku.r2dbc.mysql.MySqlResult.processRow(MySqlResult.java:176)
    at dev.miku.r2dbc.mysql.MySqlResult.handleResult(MySqlResult.java:149)
    at dev.miku.r2dbc.mysql.MySqlResult.lambda$map$1(MySqlResult.java:93)
    at reactor.core.publisher.FluxHandle$HandleSubscriber.onNext(FluxHandle.java:102)
    at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
    at dev.miku.r2dbc.mysql.util.DiscardOnCancelSubscriber.onNext(DiscardOnCancelSubscriber.java:70)
    at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drainRegular(FluxWindowPredicate.java:667)
    at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drain(FluxWindowPredicate.java:745)
    at reactor.core.publisher.FluxWindowPredicate$WindowFlux.onNext(FluxWindowPredicate.java:787)
    at reactor.core.publisher.FluxWindowPredicate$WindowPredicateMain.onNext(FluxWindowPredicate.java:265)
    at reactor.core.publisher.FluxHandleFuseable$HandleFuseableSubscriber.onNext(FluxHandleFuseable.java:184)
    at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
    at dev.miku.r2dbc.mysql.util.DiscardOnCancelSubscriber.onNext(DiscardOnCancelSubscriber.java:70)
    at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:854)
    at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:250)
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:199)
    at reactor.core.publisher.FluxHandle$HandleSubscriber.onNext(FluxHandle.java:118)
    at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:854)
    at reactor.core.publisher.EmitterProcessor.drain(EmitterProcessor.java:491)
    at reactor.core.publisher.EmitterProcessor.tryEmitNext(EmitterProcessor.java:299)
    at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27)
    at reactor.core.publisher.EmitterProcessor.onNext(EmitterProcessor.java:265)
    at dev.miku.r2dbc.mysql.client.ReactorNettyClient$ResponseSink.next(ReactorNettyClient.java:340)
    at dev.miku.r2dbc.mysql.client.ReactorNettyClient.lambda$new$0(ReactorNettyClient.java:103)
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:184)
    at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:280)
    at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:389)
    at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:405)
    at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:94)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    at dev.miku.r2dbc.mysql.client.MessageDuplexCodec.handleDecoded(MessageDuplexCodec.java:187)
    at dev.miku.r2dbc.mysql.client.MessageDuplexCodec.channelRead(MessageDuplexCodec.java:95)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:311)
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:432)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1368)
    at io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1234)
    at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1280)
    at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:507)
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:446)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.base/java.lang.Thread.run(Thread.java:834)

If I don't create a R2dbcCustomConversions bean (as described in the guide) then I don't get this error but JSON doesn't work...

Is there another way to use JSON in my scenario?

Answer 1

For me, the AbstractR2dbcConfiguration extension implementation in the post above mentioned didn't work. Instead, I used this one and configuring the read and write converter same as before.

In my case, I'm using mysql DB and the type used as source was String just like

Answer 2

Answering my own question - Looks like some sort of a bug in spring-data-r2dbc:

In R2dbcDataAutoConfiguration there is this bean definition:

@Bean
@ConditionalOnMissingBean
public R2dbcCustomConversions r2dbcCustomConversions() {
    List<Object> converters = new ArrayList<>(this.dialect.getConverters());
    converters.addAll(R2dbcCustomConversions.STORE_CONVERTERS);
    return new R2dbcCustomConversions(
            CustomConversions.StoreConversions.of(this.dialect.getSimpleTypeHolder(), converters),
            Collections.emptyList());
}

This is the bean we override when registering our converters (this is my code):

@Bean
public R2dbcCustomConversions r2dbcCustomConversions(JsonEventsReadingConverter readingConverter,
                                                     JsonEventsWritingConverter writingConverter) {
    val converters = new ArrayList<>();
    converters.add(readingConverter);
    converters.add(writingConverter);
    return new R2dbcCustomConversions(converters);
}

The CTOR new R2dbcCustomConversions(converters) takes care of adding some of the basic converters (STORE_CONVERSIONS & R2dbcConverters) but it does not add MySqlDialect.BooleanToByteConverter & MySqlDialect.ByteToBooleanConverter which takes care of the MySQL value to Java conversions.

I found an icky solution (Manually adding those converters in my bean) but I don’t see another way (Open to suggestions):

@Configuration
@RequiredArgsConstructor
public class R2dbcConvertorsConfig {
    private final DatabaseClient databaseClient;
    @Bean
    public R2dbcCustomConversions r2dbcCustomConversions(JsonEventsReadingConverter readingConverter,
                                                         JsonEventsWritingConverter writingConverter) {
        val dialect = DialectResolver.getDialect(this.databaseClient.getConnectionFactory());
        val converters = new ArrayList<>();
        converters.add(readingConverter);
        converters.add(writingConverter);
        converters.addAll(dialect.getConverters());
        return new R2dbcCustomConversions(converters);
    }
}

As requested:

package com.mycompany.platform.webhookdelivery.core.dal.converters;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.mycompany.platform.webhookdelivery.core.dal.entities.webhookfilters.JsonEvents;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.serializer.support.SerializationFailedException;
import org.springframework.data.convert.ReadingConverter;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Slf4j
@ReadingConverter
@AllArgsConstructor
@Component
public class JsonEventsReadingConverter implements Converter<String, JsonEvents> {

    private final ObjectMapper objectMapper;

    @Override
    public JsonEvents convert(@NonNull String target) {
        try {
            String json = " {\"events\":" + target + "}";
            return objectMapper.readValue(json, JsonEvents.class);
        } catch (IOException e) {
            log.debug("Failed to deserialise  JSON: " + target, e);
            throw new SerializationFailedException("Failed to deserialize  JSON: " + target, e);
        }
    }
}

package com.mycompany.platform.webhookdelivery.core.dal.converters;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mycompany.platform.webhookdelivery.core.dal.entities.webhookfilters.JsonEvents;
import lombok.AllArgsConstructor;
import lombok.extern.log4j.Log4j2;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.serializer.support.SerializationFailedException;
import org.springframework.data.convert.WritingConverter;
import org.springframework.stereotype.Component;

@Log4j2
@WritingConverter
@AllArgsConstructor
@Component
public class JsonEventsWritingConverter implements Converter<JsonEvents, String> {

    private final ObjectMapper objectMapper;

    @Override
    public String convert(JsonEvents source) {

        try {
            return objectMapper.writeValueAsString(source.getEvents());
        } catch (JsonProcessingException e) {
            log.debug("Failed to serialise JSON: " + source, e);
            throw new SerializationFailedException("Failed to serialize JSON: " + source, e);
        }
    }
}