Custom Rx Retryable logic to call Client fails when Kafka dependencies are added

[jacqui932]

There is a conflict between custom Rx logic to retry calling a Client and Kafka test dependencies.

This affects version 2.0.0 of Micronaut.

I have tracked it down to the Scala dependency which is pulled in by the Kafka test dependencies.

Tested on Windows and Mac OSX, Java 1.8 and Java 11

Steps to reproduce:

Check out this repo:

https://github.com/jacqui932/retryable-kafka-issue

Run ./gradlew test - Build will fail
Go into build.gradle and comment out Kafka dependencies in testImplementation
Run ./gradlew test - Build will pass

The custom retry logic is in the IdentityService.java class.

This project is set up to send 2 Server errors from a controller and then after that all success messages.

Expected:

2 errors are in the log about 'Server Error' and then success

Actual:

Retry loop gets stuck -

{"@timestamp":"2020-07-26T17:08:19.122+01:00","error_message":"Internal Server Error","trace_id":"identity-service-1595779698665","error_type":"io.micronaut.http.client.exceptions.HttpClientResponseException","type":"identity","logger_name":"myProjectLog","level":"ERROR","level_value":40000,"stack_trace":"<#9fd13dde> io.micronaut.http.client.exceptions.HttpClientResponseException: Internal Server Error\n\tat io.micronaut.http.client.netty.DefaultHttpClient$11.channelRead0(DefaultHttpClient.java:2046)\n\tat io.micronaut.http.client.netty.DefaultHttpClient$11.channelRead0(DefaultHttpClient.java:1964)\n\tat io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)\n\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)\n\tat io.micronaut.http.netty.stream.HttpStreamsHandler.channelRead(HttpStreamsHandler.java:190)\n\tat io.micronaut.http.netty.stream.HttpStreamsClientHandler.channelRead(HttpStreamsClientHandler.java:185)\n\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)\n\tat io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:102)\n\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)\n\tat io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:102)\n\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)\n\tat io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)\n\t... 23 frames truncated\n","message":"Error calling identity","thread_name":"nioEventLoopGroup-1-2"}
{"@timestamp":"2020-07-26T17:08:19.122+01:00","trace_id":"identity-service-1595779698665","retryCount":"1","type":"identity","logger_name":"myProjectLog","level":"WARN","level_value":30000,"message":"Retrying call to identity","thread_name":"nioEventLoopGroup-1-2"}
{"@timestamp":"2020-07-26T17:08:19.746+01:00","error_message":"Error encoding object [UnpooledHeapByteBuf(freed)] to JSON: refCnt: 0 (through reference chain: io.netty.buffer.UnpooledHeapByteBuf["array"])","trace_id":"identity-service-1595779698665","error_type":"io.micronaut.http.codec.CodecException","type":"identity","logger_name":"myProjectLog","level":"ERROR","level_value":40000,"stack_trace":"<#67464042> io.micronaut.http.codec.CodecException: Error encoding object [UnpooledHeapByteBuf(freed)] to JSON: refCnt: 0 (through reference chain: io.netty.buffer.UnpooledHeapByteBuf["array"])\n\tat io.micronaut.jackson.codec.JacksonMediaTypeCodec.encode(JacksonMediaTypeCodec.java:215)\n\tat io.micronaut.jackson.codec.JacksonMediaTypeCodec.encode(JacksonMediaTypeCodec.java:221)\n\tat io.micronaut.http.client.netty.DefaultHttpClient.lambda$buildNettyRequest$44(DefaultHttpClient.java:1573)\n\tat java.util.Optional.map(Optional.java:265)\n\tat io.micronaut.http.client.netty.DefaultHttpClient.buildNettyRequest(DefaultHttpClient.java:1573)\n\tat io.micronaut.http.client.netty.DefaultHttpClient.sendRequestThroughChannel(DefaultHttpClient.java:1747)\n\tat io.micronaut.http.client.netty.DefaultHttpClient.lambda$null$33(DefaultHttpClient.java:1110)\n\tat io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:577)\n\tat io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:570)\n\tat io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:549)\n\tat io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:490)\n\tat io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:615)\n\tat io.netty.util.concurrent.DefaultPromise.setSuccess0(DefaultPromise.java:604)\n\tat io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:104)\n\tat io.netty.channel.DefaultChannelPromise.trySuccess(DefaultChannelPromise.java:84)\n\tat io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.fulfillConnectPromise(AbstractNioChannel.java:300)\n\tat io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:335)\n\tat io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:702)\n\tat io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:650)\n\tat io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:576)\n\t... 5 frames truncated\nCaused by:

Second error repeats until time out.

[ojebari-ma]

Can you please confirm that this is still an issue in the latest version of Micronaut?

[jacquicollier]

I have checked out the project, updated the code to the latest version of Micronaut and I can confirm that this is no longer an issue and so it can be closed.