Initialize TLS context only when attempting connection (#4399)

* Initialize TLS context only when attempting connection

* Fixed eviction time check

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.client.impl;
 
 import java.security.cert.X509Certificate;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
 import io.netty.channel.ChannelInitializer;
@@ -28,9 +29,9 @@ import io.netty.handler.ssl.SslContext;
 
 import org.apache.pulsar.client.api.AuthenticationDataProvider;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.util.ObjectCache;
 import org.apache.pulsar.common.api.ByteBufPair;
 import org.apache.pulsar.common.api.Commands;
-import org.apache.pulsar.common.api.PulsarDecoder;
 import org.apache.pulsar.common.util.SecurityUtility;
 
 public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel> {
@@ -38,44 +39,54 @@ public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel>
     public static final String TLS_HANDLER = "tls";
 
     private final Supplier<ClientCnx> clientCnxSupplier;
-    private final SslContext sslCtx;
-    private final ClientConfigurationData conf;
+    private final boolean tlsEnabled;
+
+    private final Supplier<SslContext> sslContextSupplier;
+
+    private static final long TLS_CERTIFICATE_CACHE_MILLIS = TimeUnit.MINUTES.toMillis(1);
 
     public PulsarChannelInitializer(ClientConfigurationData conf, Supplier<ClientCnx> clientCnxSupplier)
             throws Exception {
         super();
         this.clientCnxSupplier = clientCnxSupplier;
+        this.tlsEnabled = conf.isUseTls();
+
         if (conf.isUseTls()) {
-            // Set client certificate if available
-            AuthenticationDataProvider authData = conf.getAuthentication().getAuthData();
-            if (authData.hasDataForTls()) {
-                this.sslCtx = SecurityUtility.createNettySslContextForClient(conf.isTlsAllowInsecureConnection(),
-                        conf.getTlsTrustCertsFilePath(), (X509Certificate[]) authData.getTlsCertificates(),
-                        authData.getTlsPrivateKey());
-            } else {
-                this.sslCtx = SecurityUtility.createNettySslContextForClient(conf.isTlsAllowInsecureConnection(),
-                        conf.getTlsTrustCertsFilePath());
-            }
+            sslContextSupplier = new ObjectCache<SslContext>(() -> {
+                try {
+                    // Set client certificate if available
+                    AuthenticationDataProvider authData = conf.getAuthentication().getAuthData();
+                    if (authData.hasDataForTls()) {
+                        return SecurityUtility.createNettySslContextForClient(conf.isTlsAllowInsecureConnection(),
+                                conf.getTlsTrustCertsFilePath(), (X509Certificate[]) authData.getTlsCertificates(),
+                                authData.getTlsPrivateKey());
+                    } else {
+                        return SecurityUtility.createNettySslContextForClient(conf.isTlsAllowInsecureConnection(),
+                                conf.getTlsTrustCertsFilePath());
+                    }
+                } catch (Exception e) {
+                    throw new RuntimeException("Failed to create TLS context", e);
+                }
+            }, TLS_CERTIFICATE_CACHE_MILLIS, TimeUnit.MILLISECONDS);
         } else {
-            this.sslCtx = null;
+            sslContextSupplier = null;
         }
-        this.conf = conf;
     }
 
     @Override
     public void initChannel(SocketChannel ch) throws Exception {
-        if (sslCtx != null) {
-            ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
+        if (tlsEnabled) {
+            ch.pipeline().addLast(TLS_HANDLER, sslContextSupplier.get().newHandler(ch.alloc()));
             ch.pipeline().addLast("ByteBufPairEncoder", ByteBufPair.COPYING_ENCODER);
         } else {
             ch.pipeline().addLast("ByteBufPairEncoder", ByteBufPair.ENCODER);
         }
 
         ch.pipeline()
-          .addLast("frameDecoder",
-                   new LengthFieldBasedFrameDecoder(
-                       Commands.DEFAULT_MAX_MESSAGE_SIZE + Commands.MESSAGE_SIZE_FRAME_PADDING,
-                       0, 4, 0, 4));
+                .addLast("frameDecoder",
+                        new LengthFieldBasedFrameDecoder(
+                                Commands.DEFAULT_MAX_MESSAGE_SIZE + Commands.MESSAGE_SIZE_FRAME_PADDING,
+                                0, 4, 0, 4));
         ch.pipeline().addLast("handler", clientCnxSupplier.get());
     }
 }

pulsar-client/src/main/java/org/apache/pulsar/client/util/ObjectCache.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.util;
+
+import java.util.function.Supplier;
+import java.time.Clock;
+import java.util.concurrent.TimeUnit;
+
+public class ObjectCache<T> implements Supplier<T> {
+
+    private final Supplier<T> supplier;
+    private T cachedInstance;
+
+    private final long cacheDurationMillis;
+    private long lastRefreshTimestamp;
+    private final Clock clock;
+
+    public ObjectCache(Supplier<T> supplier, long cacheDuration, TimeUnit unit) {
+        this(supplier, cacheDuration, unit, Clock.systemUTC());
+    }
+
+    ObjectCache(Supplier<T> supplier, long cacheDuration, TimeUnit unit, Clock clock) {
+        this.supplier = supplier;
+        this.cachedInstance = null;
+        this.cacheDurationMillis = unit.toMillis(cacheDuration);
+        this.clock = clock;
+    }
+
+    public synchronized T get() {
+        long now = clock.millis();
+        if (cachedInstance == null || (now - lastRefreshTimestamp) >= cacheDurationMillis) {
+            cachedInstance = supplier.get();
+            lastRefreshTimestamp = now;
+        }
+
+        return cachedInstance;
+    }
+}

pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientInitializationTest.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.*;
+
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.testng.annotations.Test;
+
+public class ClientInitializationTest {
+
+    @Test
+    public void testInitializeAuthWithTls() throws PulsarClientException {
+        Authentication auth = mock(Authentication.class);
+
+        PulsarClient.builder()
+                .serviceUrl("pulsar+ssl://my-host:6650")
+                .authentication(auth)
+                .build();
+
+        // Auth should only be started, though we shouldn't have tried to get credentials yet (until we first attempt to
+        // connect).
+        verify(auth).start();
+        verifyNoMoreInteractions(auth);
+    }
+}

pulsar-client/src/test/java/org/apache/pulsar/client/util/ObjectCacheTest.java

+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.util;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+
+import java.time.Clock;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
+
+import org.testng.annotations.Test;
+
+public class ObjectCacheTest {
+    @Test
+    public void testCache() {
+
+        AtomicLong currentTime = new AtomicLong(0);
+
+        Clock clock = mock(Clock.class);
+        when(clock.millis()).then(invocation -> currentTime);
+
+        AtomicInteger currentValue = new AtomicInteger(0);
+
+        Supplier<Integer> cache = new ObjectCache<>(() -> currentValue.getAndIncrement(),
+                10, TimeUnit.MILLISECONDS, clock);
+
+        assertEquals(cache.get().intValue(), 0);
+        assertEquals(cache.get().intValue(), 0);
+
+        currentTime.set(1);
+        // Still the value has not expired
+        assertEquals(cache.get().intValue(), 0);
+
+        currentTime.set(10);
+        assertEquals(cache.get().intValue(), 1);
+
+
+        currentTime.set(15);
+        assertEquals(cache.get().intValue(), 1);
+
+        currentTime.set(22);
+        assertEquals(cache.get().intValue(), 2);
+    }
+}