IGNITE-19007 Java client: Improve logging

modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java

@@ -166,6 +166,10 @@ public List<ClusterNode> connections() {
         return res;
     }
 
+    public IgniteClientConfiguration configuration() {
+        return clientCfg;
+    }
+
     /**
      * Sends request and handles response asynchronously.
      *
@@ -560,7 +564,19 @@ private CompletableFuture<ClientChannel> getCurChannelAsync() {
     private boolean shouldRetry(int opCode, ClientFutureUtils.RetryContext ctx) {
         ClientOperationType opType = ClientUtils.opCodeToClientOperationType(opCode);
 
-        return shouldRetry(opType, ctx);
+        boolean res = shouldRetry(opType, ctx);
+
+        if (log.isDebugEnabled()) {
+            if (res) {
+                log.debug("Retrying operation [opCode=" + opCode + ", opType=" + opType + ", attempt=" + ctx.attempt
+                        + ", lastError=" + ctx.lastError() + ']');
+            } else {
+                log.debug("Not retrying operation [opCode=" + opCode + ", opType=" + opType + ", attempt=" + ctx.attempt
+                        + ", lastError=" + ctx.lastError() + ']');
+            }
+        }
+
+        return res;
     }
 
     /** Determines whether specified operation should be retried. */
@@ -596,14 +612,7 @@ private boolean shouldRetry(@Nullable ClientOperationType opType, ClientFutureUt
         RetryPolicyContext retryPolicyContext = new RetryPolicyContextImpl(clientCfg, opType, ctx.attempt, exception);
 
         // Exception in shouldRetry will be handled by ClientFutureUtils.doWithRetryAsync
-        boolean shouldRetry = plc.shouldRetry(retryPolicyContext);
-
-        if (shouldRetry) {
-            log.debug("Going to retry operation because of error [op={}, currentAttempt={}, errMsg={}]",
-                    exception, opType, ctx.attempt, exception.getMessage());
-        }
-
-        return shouldRetry;
+        return plc.shouldRetry(retryPolicyContext);
     }
 
     /**

modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java

@@ -137,6 +137,10 @@ private CompletableFuture<ClientChannel> initAsync(ClientConnectionMultiplexer c
         return connMgr
                 .openAsync(cfg.getAddress(), this, this)
                 .thenCompose(s -> {
+                    if (log.isDebugEnabled()) {
+                        log.debug("Connection established [remoteAddress=" + s.remoteAddress() + ']');
+                    }
+
                     sock = s;
 
                     return handshakeAsync(DEFAULT_VERSION);
@@ -207,7 +211,10 @@ public void onMessage(ByteBuf buf) {
     /** {@inheritDoc} */
     @Override
     public void onDisconnected(@Nullable Exception e) {
-        log.debug("Disconnected from server: " + cfg.getAddress());
+        if (log.isDebugEnabled()) {
+            log.debug("Connection closed [remoteAddress=" + cfg.getAddress() + ']');
+        }
+
         close(e);
     }
 
@@ -219,6 +226,10 @@ public <T> CompletableFuture<T> serviceAsync(
             PayloadReader<T> payloadReader
     ) {
         try {
+            if (log.isTraceEnabled()) {
+                log.trace("Sending request [opCode=" + opCode + ", remoteAddress=" + cfg.getAddress() + ']');
+            }
+
             ClientRequestFuture fut = send(opCode, payloadWriter);
 
             return receiveAsync(fut, payloadReader);
@@ -268,6 +279,9 @@ private ClientRequestFuture send(int opCode, PayloadWriter payloadWriter) {
 
             return fut;
         } catch (Throwable t) {
+            log.warn("Failed to send request [id=" + id + ", op=" + opCode + ", remoteAddress=" + cfg.getAddress() + "]: "
+                    + t.getMessage(), t);
+
             // Close buffer manually on fail. Successful write closes the buffer automatically.
             payloadCh.close();
             pendingReqs.remove(id);
@@ -297,6 +311,8 @@ private <T> CompletableFuture<T> receiveAsync(ClientRequestFuture pendingReq, Pa
             try (var in = new PayloadInputChannel(this, payload)) {
                 return payloadReader.apply(in);
             } catch (Exception e) {
+                log.error("Failed to deserialize server response [remoteAddress=" + cfg.getAddress() + "]: " + e.getMessage(), e);
+
                 throw new IgniteClientConnectionException(PROTOCOL_ERR, "Failed to deserialize server response: " + e.getMessage(), e);
             }
         }, asyncContinuationExecutor);
@@ -317,6 +333,8 @@ private void processNextMessage(ByteBuf buf) throws IgniteException {
         var type = unpacker.unpackInt();
 
         if (type != ServerMessageType.RESPONSE) {
+            log.error("Unexpected message type [remoteAddress=" + cfg.getAddress() + "]: " + type);
+
             throw new IgniteClientConnectionException(PROTOCOL_ERR, "Unexpected message type: " + type);
         }
 
@@ -325,12 +343,18 @@ private void processNextMessage(ByteBuf buf) throws IgniteException {
         ClientRequestFuture pendingReq = pendingReqs.remove(resId);
 
         if (pendingReq == null) {
+            log.error("Unexpected response ID [remoteAddress=" + cfg.getAddress() + "]: " + resId);
+
             throw new IgniteClientConnectionException(PROTOCOL_ERR, String.format("Unexpected response ID [%s]", resId));
         }
 
         int flags = unpacker.unpackInt();
 
         if (ResponseFlags.getPartitionAssignmentChangedFlag(flags)) {
+            if (log.isInfoEnabled()) {
+                log.info("Partition assignment change notification received [remoteAddress=" + cfg.getAddress() + "]");
+            }
+
             for (Consumer<ClientChannel> listener : assignmentChangeListeners) {
                 listener.accept(this);
             }
@@ -484,6 +508,10 @@ private CompletableFuture<Void> handshakeRes(ClientMessageUnpacker unpacker, Pro
                     srvVer, ProtocolBitmaskFeature.allFeaturesAsEnumSet(), serverIdleTimeout, clusterNode, clusterId);
 
             return CompletableFuture.completedFuture(null);
+        } catch (Exception e) {
+            log.warn("Failed to handle handshake response [remoteAddress=" + cfg.getAddress() + "]: " + e.getMessage(), e);
+
+            return CompletableFuture.failedFuture(e);
         }
     }
 
@@ -563,7 +591,7 @@ private class HeartbeatTask extends TimerTask {
                                 .orTimeout(heartbeatTimeout, TimeUnit.MILLISECONDS)
                                 .exceptionally(e -> {
                                     if (e instanceof TimeoutException) {
-                                        log.warn("Heartbeat timeout, closing the channel");
+                                        log.warn("Heartbeat timeout, closing the channel [remoteAddress=" + cfg.getAddress() + ']');
 
                                         close((TimeoutException) e);
                                     }
@@ -572,8 +600,8 @@ private class HeartbeatTask extends TimerTask {
                                 });
                     }
                 }
-            } catch (Throwable ignored) {
-                // Ignore failed heartbeats.
+            } catch (Throwable e) {
+                log.warn("Failed to send heartbeat [remoteAddress=" + cfg.getAddress() + "]: " + e.getMessage(), e);
             }
         }
     }

modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java

@@ -31,11 +31,13 @@
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import org.apache.ignite.internal.client.ClientChannel;
+import org.apache.ignite.internal.client.ClientUtils;
 import org.apache.ignite.internal.client.PayloadOutputChannel;
 import org.apache.ignite.internal.client.ReliableChannel;
 import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
 import org.apache.ignite.internal.client.proto.ClientOp;
 import org.apache.ignite.internal.client.tx.ClientTransaction;
+import org.apache.ignite.internal.logger.IgniteLogger;
 import org.apache.ignite.internal.tostring.IgniteToStringBuilder;
 import org.apache.ignite.lang.IgniteBiTuple;
 import org.apache.ignite.lang.IgniteException;
@@ -60,6 +62,8 @@ public class ClientTable implements Table {
 
     private final ConcurrentHashMap<Integer, ClientSchema> schemas = new ConcurrentHashMap<>();
 
+    private final IgniteLogger log;
+
     private volatile int latestSchemaVer = -1;
 
     private final Object latestSchemaLock = new Object();
@@ -83,6 +87,7 @@ public ClientTable(ReliableChannel ch, UUID id, String name) {
         this.ch = ch;
         this.id = id;
         this.name = name;
+        this.log = ClientUtils.logger(ch.configuration(), ClientTable.class);
     }
 
     /**
@@ -160,13 +165,19 @@ private CompletableFuture<ClientSchema> loadSchema(@Nullable Integer ver) {
             int schemaCnt = r.in().unpackMapHeader();
 
             if (schemaCnt == 0) {
+                log.warn("Schema not found [tableId=" + id + ", schemaVersion=" + ver + "]");
+
                 throw new IgniteException(UNEXPECTED_ERR, "Schema not found: " + ver);
             }
 
             ClientSchema last = null;
 
             for (var i = 0; i < schemaCnt; i++) {
                 last = readSchema(r.in());
+
+                if (log.isDebugEnabled()) {
+                    log.debug("Schema loaded [tableId=" + id + ", schemaVersion=" + last.version() + "]");
+                }
             }
 
             return last;

modules/client/src/test/java/org/apache/ignite/client/ClientLoggingTest.java

@@ -22,9 +22,11 @@
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.startsWith;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import org.apache.ignite.client.fakes.FakeIgnite;
 import org.apache.ignite.client.fakes.FakeIgniteTables;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
 import org.apache.ignite.internal.util.IgniteUtils;
 import org.apache.ignite.lang.LoggerFactory;
 import org.junit.jupiter.api.AfterEach;
@@ -78,6 +80,28 @@ public void loggersSetToDifferentClientsNotInterfereWithEachOther() throws Excep
         loggerFactory2.logger.entries().forEach(msg -> assertThat(msg, startsWith("client2:")));
     }
 
+    @Test
+    public void testBasicLogging() throws Exception {
+        FakeIgnite ignite = new FakeIgnite();
+        ((FakeIgniteTables) ignite.tables()).createTable("t");
+
+        server = startServer(10950, ignite);
+        server2 = startServer(10955, ignite);
+
+        var loggerFactory = new TestLoggerFactory("c");
+
+        try (var client = createClient(loggerFactory)) {
+            client.tables().tables();
+            client.tables().table("t");
+
+            assertTrue(IgniteTestUtils.waitForCondition(() -> loggerFactory.logger.entries().size() > 10, 5_000));
+
+            loggerFactory.assertLogContains("Connection established");
+            loggerFactory.assertLogContains("c:Sending request [opCode=3, remoteAddress=127.0.0.1:1095");
+            loggerFactory.assertLogContains("c:Failed to establish connection to 127.0.0.1:1095");
+        }
+    }
+
     private static TestServer startServer(int port, FakeIgnite ignite) {
         return AbstractClientTest.startServer(
                 port,

modules/client/src/test/java/org/apache/ignite/client/HeartbeatTest.java

@@ -46,8 +46,8 @@ public void testHeartbeatLongerThanIdleTimeoutCausesDisconnect() throws Exceptio
             try (var ignored = builder.build()) {
                 assertTrue(
                         IgniteTestUtils.waitForCondition(
-                                () -> loggerFactory.logger.entries().stream().anyMatch(x -> x.contains("Disconnected from server")),
-                                1000));
+                                () -> loggerFactory.logger.entries().stream().anyMatch(x -> x.contains("Connection closed")),
+                                10000));
             }
         }
     }

modules/client/src/test/java/org/apache/ignite/client/RetryPolicyTest.java

@@ -187,8 +187,8 @@ public void testRetryReadPolicyRetriesReadOperations() throws Exception {
             recView.get(null, Tuple.create().set("id", 1L));
             recView.get(null, Tuple.create().set("id", 1L));
 
-            loggerFactory.assertLogContains("Disconnected from server");
-            loggerFactory.assertLogContains("Going to retry operation because of error [op=TUPLE_GET");
+            loggerFactory.assertLogContains("Connection closed");
+            loggerFactory.assertLogContains("Retrying operation [opCode=12, opType=TUPLE_GET, attempt=0, lastError=java.util");
         }
     }