diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java b/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java index 09fb936c889..5fbe5b3bf45 100644 --- a/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java +++ b/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java @@ -166,6 +166,10 @@ public List connections() { return res; } + public IgniteClientConfiguration configuration() { + return clientCfg; + } + /** * Sends request and handles response asynchronously. * @@ -560,7 +564,19 @@ private CompletableFuture 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); } /** diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java index 1036863386b..86387df6747 100644 --- a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java +++ b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java @@ -137,6 +137,10 @@ private CompletableFuture 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 CompletableFuture serviceAsync( PayloadReader 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 CompletableFuture 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 listener : assignmentChangeListeners) { listener.accept(this); } @@ -484,6 +508,10 @@ private CompletableFuture 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); } } } diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java index d4a36701b98..91c1a18bad1 100644 --- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java +++ b/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 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,6 +165,8 @@ private CompletableFuture 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); } @@ -167,6 +174,10 @@ private CompletableFuture loadSchema(@Nullable Integer ver) { 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; diff --git a/modules/client/src/test/java/org/apache/ignite/client/ClientLoggingTest.java b/modules/client/src/test/java/org/apache/ignite/client/ClientLoggingTest.java index 71e557b8364..41635df9b85 100644 --- a/modules/client/src/test/java/org/apache/ignite/client/ClientLoggingTest.java +++ b/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, diff --git a/modules/client/src/test/java/org/apache/ignite/client/HeartbeatTest.java b/modules/client/src/test/java/org/apache/ignite/client/HeartbeatTest.java index c505deaf306..aeeb47112ee 100644 --- a/modules/client/src/test/java/org/apache/ignite/client/HeartbeatTest.java +++ b/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)); } } } diff --git a/modules/client/src/test/java/org/apache/ignite/client/RetryPolicyTest.java b/modules/client/src/test/java/org/apache/ignite/client/RetryPolicyTest.java index b154503415c..f0e9f0f415c 100644 --- a/modules/client/src/test/java/org/apache/ignite/client/RetryPolicyTest.java +++ b/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"); } }