Feat/map log

trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java

@@ -72,7 +72,10 @@ protected CompletionStage<Response> doInvoke(Request request, CompletionStage<Se
     protected ConsumerInvokerProxy<T> getInvoker(ServiceInstance instance) {
         String key = toUniqKey(instance);
         ConsumerInvokerProxy<T> result = invokerCache.get(key);
-        return Optional.ofNullable(result).orElseGet(() -> createInvoker(instance));
+        if (result != null && result.isAvailable()) {
+            return result;
+        }
+        return createInvoker(instance);
     }
 
     @SuppressWarnings("rawtypes")

trpc-core/src/main/java/com/tencent/trpc/core/common/timer/HashedWheelTimer.java

@@ -261,7 +261,7 @@ public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit
         }
 
         if (duration < MILLISECOND_NANOS) {
-            logger.warn("Configured tickDuration %d smaller then %d, using 1ms.", tickDuration,
+            logger.warn("Configured tickDuration {} smaller then {}, using 1ms.", tickDuration,
                     MILLISECOND_NANOS);
             this.tickDuration = MILLISECOND_NANOS;
         } else {

trpc-core/src/main/java/com/tencent/trpc/core/transport/AbstractClientTransport.java

@@ -178,10 +178,10 @@ public void close() {
     @Override
     public CompletionStage<Void> send(Object msg) throws TransportException {
         if (isClosed()) {
-            throw TransportException.create(String.format(
+            throw TransportException.create(
                     "Client transport(transport=%s, class=%s, msg=%s) send fail, due to transport"
                             + " is not available or close",
-                    this, name, msg));
+                    this, name, msg);
         }
         return getChannel0().thenCompose(f -> f.send(msg));
     }
@@ -192,10 +192,10 @@ public CompletionStage<Void> send(Object msg) throws TransportException {
     @Override
     public CompletionStage<Channel> getChannel() throws TransportException {
         if (isClosed()) {
-            throw TransportException.create(String.format(
+            throw TransportException.create(
                     "Client transport(transport=%s, class=%s) get channel fail, due to transport "
                             + "is not available or close",
-                    this, name));
+                    this, name);
         }
         return getChannel0();
     }
@@ -420,8 +420,7 @@ protected void stopInternal() throws Exception {
             try {
                 doClose();
             } catch (Throwable ex) {
-                logger.error(String.format("Client transport(%s) destroy failed", getRemoteAddress(),
-                        ex));
+                logger.error(String.format("Client transport(%s) destroy failed", getRemoteAddress()), ex);
             }
             try {
                 if (handler != null) {

trpc-core/src/test/java/com/tencent/trpc/core/cluster/def/DefClusterInvokerTest.java

@@ -33,10 +33,12 @@
 import com.tencent.trpc.core.worker.handler.TrpcThreadExceptionHandler;
 import com.tencent.trpc.core.worker.spi.WorkerPool;
 import com.tencent.trpc.core.worker.support.thread.ThreadWorkerPool;
+import java.lang.reflect.Field;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicLong;
@@ -204,4 +206,55 @@ public void testDoInvoke() {
         }
     }
 
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testGetInvokerSkipsUnavailableCachedInvoker() throws Exception {
+        Assertions.assertFalse(consumerInvokerProxy.isAvailable());
+
+        // Put the unavailable invoker into invokerCache via reflection
+        Field cacheField = DefClusterInvoker.class.getDeclaredField("invokerCache");
+        cacheField.setAccessible(true);
+        ConcurrentMap<String, ConsumerInvokerProxy<GenericClient>> cache =
+                (ConcurrentMap<String, ConsumerInvokerProxy<GenericClient>>) cacheField.get(defClusterInvoker);
+
+        ServiceInstance instance = new ServiceInstance("127.0.0.1", 12345);
+        String key = "127.0.0.1:12345:null";
+        cache.put(key, consumerInvokerProxy);
+
+        // Spy on defClusterInvoker and mock createInvoker to return a new available proxy
+        DefClusterInvoker<GenericClient> spy = Mockito.spy(defClusterInvoker);
+        ConsumerInvokerProxy<GenericClient> newProxy = Mockito.mock(ConsumerInvokerProxy.class);
+        Mockito.when(newProxy.isAvailable()).thenReturn(true);
+        Mockito.doReturn(newProxy).when(spy).createInvoker(instance);
+
+        ConsumerInvokerProxy<GenericClient> result = spy.getInvoker(instance);
+
+        // Should not return the unavailable cached invoker
+        Assertions.assertNotSame(consumerInvokerProxy, result);
+        // Should return the new available one from createInvoker
+        Assertions.assertSame(newProxy, result);
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testGetInvokerReturnsAvailableCachedInvoker() throws Exception {
+        // Put an available invoker into invokerCache via reflection
+        Field cacheField = DefClusterInvoker.class.getDeclaredField("invokerCache");
+        cacheField.setAccessible(true);
+        ConcurrentMap<String, ConsumerInvokerProxy<GenericClient>> cache =
+                (ConcurrentMap<String, ConsumerInvokerProxy<GenericClient>>) cacheField.get(defClusterInvoker);
+
+        ServiceInstance instance = new ServiceInstance("127.0.0.1", 12345);
+        String key = "127.0.0.1:12345:null";
+
+        // Create an available proxy (client.isAvailable() = true)
+        ConsumerInvokerProxy<GenericClient> availableProxy = Mockito.mock(ConsumerInvokerProxy.class);
+        Mockito.when(availableProxy.isAvailable()).thenReturn(true);
+        cache.put(key, availableProxy);
+
+        // getInvoker should return the cached available proxy directly
+        ConsumerInvokerProxy<GenericClient> result = defClusterInvoker.getInvoker(instance);
+        Assertions.assertSame(availableProxy, result);
+    }
+
 }

trpc-core/src/test/java/com/tencent/trpc/core/common/timer/HashWheelTimerTest.java

@@ -54,6 +54,54 @@ public void testInit() {
         hashedWheelTimer.stop();
     }
 
+    /**
+     * A tick duration smaller than one millisecond is normalized to one millisecond and a warning is logged. The
+     * warning uses the {@code {}} placeholder of slf4j, a printf style placeholder would leave the raw text in the
+     * log and lose the real values.
+     */
+    @Test
+    public void testTickDurationSmallerThanOneMillisecond() {
+        HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(new NamedThreadFactory(), 1L,
+                TimeUnit.NANOSECONDS);
+        try {
+            Assertions.assertNotNull(hashedWheelTimer);
+            hashedWheelTimer.start();
+            // the tick duration has been normalized, so the timer still works
+            Timeout timeout = hashedWheelTimer.newTimeout(t -> {
+            }, 1000, TimeUnit.MILLISECONDS);
+            Assertions.assertFalse(timeout.isExpired());
+            timeout.cancel();
+        } finally {
+            hashedWheelTimer.stop();
+        }
+    }
+
+    /**
+     * An illegal tick duration should be rejected.
+     */
+    @Test
+    public void testIllegalTickDuration() {
+        try {
+            new HashedWheelTimer(new NamedThreadFactory(), 0L, TimeUnit.MILLISECONDS);
+            Assertions.fail("IllegalArgumentException is expected");
+        } catch (IllegalArgumentException e) {
+            Assertions.assertTrue(e.getMessage().contains("tickDuration must be greater than 0"), e.getMessage());
+        }
+    }
+
+    /**
+     * A tick duration which overflows the wheel should be rejected.
+     */
+    @Test
+    public void testTickDurationOverflow() {
+        try {
+            new HashedWheelTimer(new NamedThreadFactory(), Long.MAX_VALUE / 2, TimeUnit.DAYS);
+            Assertions.fail("IllegalArgumentException is expected");
+        } catch (IllegalArgumentException e) {
+            Assertions.assertTrue(e.getMessage().contains("tickDuration"), e.getMessage());
+        }
+    }
+
     @Test
     public void test() {
         timer.toString();

trpc-core/src/test/java/com/tencent/trpc/core/exception/TransportExceptionTest.java

@@ -0,0 +1,138 @@
+/*
+ * Tencent is pleased to support the open source community by making tRPC available.
+ *
+ * Copyright (C) 2023 Tencent.
+ * All rights reserved.
+ *
+ * If you have downloaded a copy of the tRPC source code from Tencent,
+ * please note that tRPC source code is licensed under the Apache 2.0 License,
+ * A copy of the Apache 2.0 License can be found in the LICENSE file.
+ */
+
+package com.tencent.trpc.core.exception;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.UnknownFormatConversionException;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test {@link TransportException}, especially the {@code create} methods which format the message by themselves, so
+ * the callers must pass the raw format and the arguments instead of an already formatted string.
+ */
+public class TransportExceptionTest {
+
+    private static final String MESSAGE = "transport error";
+
+    @Test
+    public void testConstructWithMessage() {
+        TransportException e = new TransportException(MESSAGE);
+        assertEquals(MESSAGE, e.getMessage());
+        assertNull(e.getCause());
+    }
+
+    @Test
+    public void testConstructWithMessageAndCause() {
+        IllegalStateException cause = new IllegalStateException("cause");
+        TransportException e = new TransportException(MESSAGE, cause);
+        assertEquals(MESSAGE, e.getMessage());
+        assertSame(cause, e.getCause());
+    }
+
+    @Test
+    public void testConstructWithCause() {
+        IllegalStateException cause = new IllegalStateException("cause");
+        TransportException e = new TransportException(cause);
+        assertEquals("cause", e.getMessage());
+        assertSame(cause, e.getCause());
+    }
+
+    @Test
+    public void testCreateFormatsArguments() {
+        TransportException e = TransportException.create("send fail, addr=%s, port=%s", "127.0.0.1", 8080);
+        assertEquals("send fail, addr=127.0.0.1, port=8080", e.getMessage());
+        assertNull(e.getCause());
+    }
+
+    @Test
+    public void testCreateWithCauseFormatsArguments() {
+        IllegalStateException cause = new IllegalStateException("cause");
+        TransportException e = TransportException.create(cause, "send fail, addr=%s", "127.0.0.1");
+        assertEquals("send fail, addr=127.0.0.1", e.getMessage());
+        assertSame(cause, e.getCause());
+    }
+
+    @Test
+    public void testCreateWithNullCauseFormatsArguments() {
+        TransportException e = TransportException.create((Throwable) null, "send fail, addr=%s", "127.0.0.1");
+        assertEquals("send fail, addr=127.0.0.1", e.getMessage());
+        assertNull(e.getCause());
+    }
+
+    @Test
+    public void testCreateWithoutArguments() {
+        TransportException e = TransportException.create("send fail");
+        assertEquals("send fail", e.getMessage());
+    }
+
+    @Test
+    public void testCreateWithNullArgument() {
+        TransportException e = TransportException.create("send fail, msg=%s", (Object) null);
+        assertEquals("send fail, msg=null", e.getMessage());
+    }
+
+    /**
+     * The argument may contain a percent sign, and it must be kept as it is, because only the format is formatted.
+     */
+    @Test
+    public void testCreateKeepsPercentSignOfArgument() {
+        TransportException e = TransportException.create("send fail, msg=%s", "GET /a%2Fb?rate=100%");
+        assertEquals("send fail, msg=GET /a%2Fb?rate=100%", e.getMessage());
+    }
+
+    /**
+     * The caller must never pass an already formatted message which contains a percent sign, otherwise the message
+     * is formatted twice and the real error is hidden by a format exception. This test pins down the behaviour so
+     * that the double formatting can be caught.
+     */
+    @Test
+    public void testCreateWithAlreadyFormattedMessageContainingPercentFails() {
+        String formatted = String.format("send fail, msg=%s", "GET /a%2Fb");
+        assertEquals("send fail, msg=GET /a%2Fb", formatted);
+        try {
+            TransportException.create(formatted);
+            fail("UnknownFormatConversionException is expected");
+        } catch (UnknownFormatConversionException e) {
+            assertNotNull(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testTransReturnsTheSameInstanceForTransportException() {
+        TransportException origin = new TransportException(MESSAGE);
+        assertSame(origin, TransportException.trans(origin));
+        assertSame(origin, TransportException.trans(origin, "other message"));
+    }
+
+    @Test
+    public void testTransWrapsOtherException() {
+        IllegalStateException cause = new IllegalStateException("cause");
+        TransportException e = TransportException.trans(cause);
+        assertEquals("cause", e.getMessage());
+        assertSame(cause, e.getCause());
+
+        TransportException withMessage = TransportException.trans(cause, MESSAGE);
+        assertEquals(MESSAGE, withMessage.getMessage());
+        assertSame(cause, withMessage.getCause());
+    }
+
+    @Test
+    public void testIsRuntimeException() {
+        assertTrue(new TransportException(MESSAGE) instanceof RuntimeException);
+    }
+}