fix(logging): prevent excessive MySQL warning logs

Lower MySQL warning log level from INFO to DEBUG to reduce log spam while maintaining debugging visibility.

Closes #311

Author: jchrys

r2dbc-mysql/src/main/java/io/asyncer/r2dbc/mysql/client/ReactorNettyClient.java

@@ -89,7 +89,7 @@ final class ReactorNettyClient implements Client {
         requireNonNull(context, "context must not be null");
         requireNonNull(ssl, "ssl must not be null");
         require(responseProcessor.asFlux() instanceof Subscriber,
-            "responseProcessor(" + responseProcessor + ") must be a Subscriber");
+                "responseProcessor(" + responseProcessor + ") must be a Subscriber");
 
         this.connection = connection;
         this.context = context;
@@ -105,54 +105,54 @@ final class ReactorNettyClient implements Client {
             logger.debug("Connection tracking logging is enabled");
 
             connection.addHandlerFirst(LoggingHandler.class.getSimpleName(),
-                new LoggingHandler(ReactorNettyClient.class, LogLevel.TRACE));
+                                       new LoggingHandler(ReactorNettyClient.class, LogLevel.TRACE));
         }
 
         ResponseSink sink = new ResponseSink();
 
         connection.inbound().receiveObject()
-            .doOnNext(it -> {
-                if (it instanceof ServerMessage) {
-                    if (it instanceof ReferenceCounted) {
-                        ((ReferenceCounted) it).retain();
-                    }
-                    sink.next((ServerMessage) it);
-                } else {
-                    // ReferenceCounted will be released by Netty.
-                    throw ClientExceptions.unsupportedProtocol(it.getClass().getTypeName());
-                }
-            })
-            .onErrorResume(this::resumeError)
-            .subscribe(new ResponseSubscriber(sink));
+                  .doOnNext(it -> {
+                      if (it instanceof ServerMessage) {
+                          if (it instanceof ReferenceCounted) {
+                              ((ReferenceCounted) it).retain();
+                          }
+                          sink.next((ServerMessage) it);
+                      } else {
+                          // ReferenceCounted will be released by Netty.
+                          throw ClientExceptions.unsupportedProtocol(it.getClass().getTypeName());
+                      }
+                  })
+                  .onErrorResume(this::resumeError)
+                  .subscribe(new ResponseSubscriber(sink));
 
         this.requests.asFlux()
-            .concatMap(message -> {
-                if (DEBUG_ENABLED) {
-                    logger.debug("Request: {}", message);
-                }
-
-                if (message == ExitMessage.INSTANCE) {
-                    if (STATE_UPDATER.compareAndSet(this, ST_CONNECTED, ST_CLOSING)) {
-                        logger.debug("Exit message sent");
-                    } else {
-                        logger.debug("Exit message sent (duplicated / connection already closed)");
-                    }
-                }
-
-                if (message.isSequenceReset()) {
-                    resetSequence(connection);
-                }
-
-                return connection.outbound().sendObject(message);
-            })
-            .onErrorResume(this::resumeError)
-            .doAfterTerminate(this::handleClose)
-            .subscribe();
+                     .concatMap(message -> {
+                         if (DEBUG_ENABLED) {
+                             logger.debug("Request: {}", message);
+                         }
+
+                         if (message == ExitMessage.INSTANCE) {
+                             if (STATE_UPDATER.compareAndSet(this, ST_CONNECTED, ST_CLOSING)) {
+                                 logger.debug("Exit message sent");
+                             } else {
+                                 logger.debug("Exit message sent (duplicated / connection already closed)");
+                             }
+                         }
+
+                         if (message.isSequenceReset()) {
+                             resetSequence(connection);
+                         }
+
+                         return connection.outbound().sendObject(message);
+                     })
+                     .onErrorResume(this::resumeError)
+                     .doAfterTerminate(this::handleClose)
+                     .subscribe();
     }
 
     @Override
     public <T> Flux<T> exchange(ClientMessage request,
-        BiConsumer<ServerMessage, SynchronousSink<T>> handler) {
+                                BiConsumer<ServerMessage, SynchronousSink<T>> handler) {
         requireNonNull(request, "request must not be null");
 
         return Mono.<Flux<T>>create(sink -> {
@@ -166,9 +166,9 @@ public <T> Flux<T> exchange(ClientMessage request,
 
             Flux<T> responses = OperatorUtils.discardOnCancel(
                 responseProcessor.asFlux()
-                    .doOnSubscribe(ignored -> emitNextRequest(request))
-                    .handle(handler)
-                    .doOnTerminate(requestQueue)
+                                 .doOnSubscribe(ignored -> emitNextRequest(request))
+                                 .handle(handler)
+                                 .doOnTerminate(requestQueue)
             ).doOnDiscard(ReferenceCounted.class, ReferenceCounted::release);
 
             requestQueue.submit(RequestTask.wrap(request, sink, responses));
@@ -203,8 +203,9 @@ public <T> Flux<T> exchange(FluxExchangeable<T> exchangeable) {
                 });
 
             requestQueue.submit(RequestTask.wrap(exchangeable, sink, OperatorUtils.discardOnCancel(responses)
-                .doOnDiscard(ReferenceCounted.class, ReferenceCounted::release)
-                .doOnCancel(exchangeable::dispose)));
+                                                                                  .doOnDiscard(ReferenceCounted.class,
+                                                                                               ReferenceCounted::release)
+                                                                                  .doOnCancel(exchangeable::dispose)));
         }).flatMapMany(Function.identity());
     }
 
@@ -272,7 +273,7 @@ private static void resetSequence(Connection connection) {
     @Override
     public String toString() {
         return String.format("ReactorNettyClient(%s){connectionId=%d}",
-            isConnected() ? "activating" : "closing or closed", context.getConnectionId());
+                             isConnected()? "activating" : "closing or closed", context.getConnectionId());
     }
 
     private void emitNextRequest(ClientMessage request) {
@@ -379,17 +380,17 @@ public void error(Throwable e) {
 
         @Override
         public void next(ServerMessage message) {
-            if (message instanceof WarningMessage) {
-                int warnings = ((WarningMessage) message).getWarnings();
-                if (warnings == 0) {
-                    if (DEBUG_ENABLED) {
+            if (DEBUG_ENABLED) {
+                if (message instanceof WarningMessage) {
+                    int warnings = ((WarningMessage) message).getWarnings();
+                    if (warnings == 0) {
                         logger.debug("Response: {}", message);
+                    } else {
+                        logger.debug("Response: {}, reports {} warning(s)", message, warnings);
                     }
-                } else if (INFO_ENABLED) {
-                    logger.info("Response: {}, reports {} warning(s)", message, warnings);
+                } else {
+                    logger.debug("Response: {}", message);
                 }
-            } else if (DEBUG_ENABLED) {
-                logger.debug("Response: {}", message);
             }
 
             responseProcessor.emitNext(message, EmitFailureHandler.FAIL_FAST);