Merge branch '1.7' into 2.0

Author: Zhen Li

driver/src/main/java/org/neo4j/driver/internal/BoltServerAddress.java

@@ -23,11 +23,15 @@
 import java.net.SocketAddress;
 import java.net.URI;
 import java.net.UnknownHostException;
+import java.util.List;
 import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.neo4j.driver.net.ServerAddress;
 
 import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.toList;
 
 /**
  * Holds a host and port pair that denotes a Bolt server address.
@@ -37,11 +41,12 @@ public class BoltServerAddress implements ServerAddress
     public static final int DEFAULT_PORT = 7687;
     public static final BoltServerAddress LOCAL_DEFAULT = new BoltServerAddress( "localhost", DEFAULT_PORT );
 
-    private final String originalHost; // This keeps the original host name provided by the user.
     private final String host; // This could either be the same as originalHost or it is an IP address resolved from the original host.
     private final int port;
     private final String stringValue;
 
+    private InetAddress resolved;
+
     public BoltServerAddress( String address )
     {
         this( uriFrom( address ) );
@@ -54,15 +59,15 @@ public BoltServerAddress( URI uri )
 
     public BoltServerAddress( String host, int port )
     {
-        this( host, host, port );
+        this( host, null, port );
     }
 
-    public BoltServerAddress( String originalHost, String host, int port )
+    private BoltServerAddress( String host, InetAddress resolved, int port )
     {
-        this.originalHost = requireNonNull( originalHost, "original host" );
         this.host = requireNonNull( host, "host" );
+        this.resolved = resolved;
         this.port = requireValidPort( port );
-        this.stringValue = String.format( "%s:%d", host, port );
+        this.stringValue = resolved != null ? String.format( "%s(%s):%d", host, resolved.getHostAddress(), port ) : String.format( "%s:%d", host, port );
     }
 
     public static BoltServerAddress from( ServerAddress address )
@@ -84,13 +89,13 @@ public boolean equals( Object o )
             return false;
         }
         BoltServerAddress that = (BoltServerAddress) o;
-        return port == that.port && originalHost.equals( that.originalHost ) && host.equals( that.host );
+        return port == that.port && host.equals( that.host );
     }
 
     @Override
     public int hashCode()
     {
-        return Objects.hash( originalHost, host, port );
+        return Objects.hash( host, port );
     }
 
     @Override
@@ -108,38 +113,39 @@ public String toString()
      */
     public SocketAddress toSocketAddress()
     {
-        return new InetSocketAddress( host, port );
+        return resolved == null ? new InetSocketAddress( host, port ) : new InetSocketAddress( resolved, port );
     }
 
     /**
-     * Resolve the host name down to an IP address, if not already resolved.
+     * Resolve the host name down to an IP address
      *
-     * @return this instance if already resolved, otherwise a new address instance
+     * @return a new address instance
      * @throws UnknownHostException if no IP address for the host could be found
      * @see InetAddress#getByName(String)
      */
     public BoltServerAddress resolve() throws UnknownHostException
     {
-        String ipAddress = InetAddress.getByName( host ).getHostAddress();
-        if ( ipAddress.equals( host ) )
-        {
-            return this;
-        }
-        else
-        {
-            return new BoltServerAddress( host, ipAddress, port );
-        }
+        return new BoltServerAddress( host, InetAddress.getByName( host ), port );
     }
 
-    @Override
-    public String host()
+    /**
+     * Resolve the host name down to all IP addresses that can be resolved to
+     *
+     * @return an array of new address instances that holds resolved addresses
+     * @throws UnknownHostException if no IP address for the host could be found
+     * @see InetAddress#getAllByName(String)
+     */
+    public List<BoltServerAddress> resolveAll() throws UnknownHostException
     {
-        return host;
+        return Stream.of( InetAddress.getAllByName( host ) )
+                .map( address -> new BoltServerAddress( host, address, port ) )
+                .collect( toList() );
     }
 
-    public String originalHost()
+    @Override
+    public String host()
     {
-        return originalHost;
+        return host;
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/DriverFactory.java

@@ -32,7 +32,6 @@
 import org.neo4j.driver.internal.async.connection.ChannelConnectorImpl;
 import org.neo4j.driver.internal.async.pool.ConnectionPoolImpl;
 import org.neo4j.driver.internal.async.pool.PoolSettings;
-import org.neo4j.driver.internal.cluster.DnsResolver;
 import org.neo4j.driver.internal.cluster.RoutingContext;
 import org.neo4j.driver.internal.cluster.RoutingSettings;
 import org.neo4j.driver.internal.cluster.loadbalancing.LeastConnectedLoadBalancingStrategy;
@@ -62,7 +61,9 @@
 
 import static java.lang.String.format;
 import static org.neo4j.driver.internal.metrics.MetricsProvider.METRICS_DISABLED_PROVIDER;
+import static org.neo4j.driver.internal.cluster.IdentityResolver.IDENTITY_RESOLVER;
 import static org.neo4j.driver.internal.security.SecurityPlan.insecure;
+import static org.neo4j.driver.internal.util.ErrorUtil.addSuppressed;
 
 public class DriverFactory
 {
@@ -245,7 +246,7 @@ private static LoadBalancingStrategy createLoadBalancingStrategy( Config config,
     private static ServerAddressResolver createResolver( Config config )
     {
         ServerAddressResolver configuredResolver = config.resolver();
-        return configuredResolver != null ? configuredResolver : new DnsResolver( config.logging() );
+        return configuredResolver != null ? configuredResolver : IDENTITY_RESOLVER;
     }
 
     /**
@@ -330,12 +331,12 @@ private static SecurityPlan createSecurityPlanImpl( BoltServerAddress address, C
             case TRUST_ON_FIRST_USE:
                 logger.warn(
                         "Option `TRUST_ON_FIRST_USE` has been deprecated and will be removed in a future " +
-                        "version of the driver. Please switch to use `TRUST_ALL_CERTIFICATES` instead." );
+                                "version of the driver. Please switch to use `TRUST_ALL_CERTIFICATES` instead." );
                 return SecurityPlan.forTrustOnFirstUse( trustStrategy.certFile(), hostnameVerificationEnabled, address, logger );
             case TRUST_SIGNED_CERTIFICATES:
                 logger.warn(
                         "Option `TRUST_SIGNED_CERTIFICATE` has been deprecated and will be removed in a future " +
-                        "version of the driver. Please switch to use `TRUST_CUSTOM_CA_SIGNED_CERTIFICATES` instead." );
+                                "version of the driver. Please switch to use `TRUST_CUSTOM_CA_SIGNED_CERTIFICATES` instead." );
                 // intentional fallthrough
 
             case TRUST_CUSTOM_CA_SIGNED_CERTIFICATES:
@@ -397,10 +398,7 @@ private static void closeConnectionPoolAndSuppressError( ConnectionPool connecti
         }
         catch ( Throwable closeError )
         {
-            if ( mainError != closeError )
-            {
-                mainError.addSuppressed( closeError );
-            }
+            addSuppressed( mainError, closeError );
         }
     }
 

driver/src/main/java/org/neo4j/driver/internal/async/connection/NettyChannelInitializer.java

@@ -77,7 +77,7 @@ private SslHandler createSslHandler()
     private SSLEngine createSslEngine()
     {
         SSLContext sslContext = securityPlan.sslContext();
-        SSLEngine sslEngine = sslContext.createSSLEngine( address.originalHost(), address.port() );
+        SSLEngine sslEngine = sslContext.createSSLEngine( address.host(), address.port() );
         sslEngine.setUseClientMode( true );
         if ( securityPlan.requiresHostnameVerification() )
         {

driver/src/main/java/org/neo4j/driver/internal/async/inbound/ChannelErrorHandler.java

@@ -88,15 +88,15 @@ public void exceptionCaught( ChannelHandlerContext ctx, Throwable error )
         else
         {
             failed = true;
-            log.error( "Fatal error occurred in the pipeline", error );
+            log.warn( "Fatal error occurred in the pipeline", error );
             fail( ctx, error );
         }
     }
 
     private void fail( ChannelHandlerContext ctx, Throwable error )
     {
         Throwable cause = transformError( error );
-        messageDispatcher.handleFatalError( cause );
+        messageDispatcher.handleChannelError( cause );
         log.debug( "Closing channel because of a failure '%s'", error );
         ctx.close();
     }

driver/src/main/java/org/neo4j/driver/internal/async/inbound/InboundMessageDispatcher.java

@@ -37,6 +37,7 @@
 
 import static java.util.Objects.requireNonNull;
 import static org.neo4j.driver.internal.messaging.request.ResetMessage.RESET;
+import static org.neo4j.driver.internal.util.ErrorUtil.addSuppressed;
 
 public class InboundMessageDispatcher implements ResponseMessageHandler
 {
@@ -141,9 +142,17 @@ public void handleIgnoredMessage()
         handler.onFailure( error );
     }
 
-    public void handleFatalError( Throwable error )
+    public void handleChannelError( Throwable error )
     {
-        currentError = error;
+        if ( currentError != null )
+        {
+            // we already have an error, this new error probably is caused by the existing one, thus we chain the new error on this current error
+            addSuppressed( currentError, error );
+        }
+        else
+        {
+            currentError = error;
+        }
         fatalErrorOccurred = true;
 
         while ( !handlers.isEmpty() )

driver/src/main/java/org/neo4j/driver/internal/async/pool/ConnectionPoolImpl.java

@@ -34,6 +34,10 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.neo4j.driver.Logger;
+import org.neo4j.driver.Logging;
+import org.neo4j.driver.exceptions.ClientException;
+import org.neo4j.driver.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.internal.BoltServerAddress;
 import org.neo4j.driver.internal.async.connection.ChannelConnector;
 import org.neo4j.driver.internal.async.connection.DirectConnection;
@@ -43,9 +47,8 @@
 import org.neo4j.driver.internal.spi.ConnectionPool;
 import org.neo4j.driver.internal.util.Clock;
 import org.neo4j.driver.internal.util.Futures;
-import org.neo4j.driver.Logger;
-import org.neo4j.driver.Logging;
-import org.neo4j.driver.exceptions.ClientException;
+
+import static java.lang.String.format;
 
 public class ConnectionPoolImpl implements ConnectionPool
 {
@@ -59,11 +62,12 @@ public class ConnectionPoolImpl implements ConnectionPool
     private final MetricsListener metricsListener;
     private final boolean ownsEventLoopGroup;
 
-    private final ConcurrentMap<BoltServerAddress,ChannelPool> pools = new ConcurrentHashMap<>();
+    private final ConcurrentMap<BoltServerAddress,ExtendedChannelPool> pools = new ConcurrentHashMap<>();
     private final AtomicBoolean closed = new AtomicBoolean();
 
     public ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, PoolSettings settings, MetricsListener metricsListener, Logging logging, Clock clock, boolean ownsEventLoopGroup )
     {
+
         this( connector, bootstrap, new NettyChannelTracker( metricsListener, bootstrap.config().group().next(), logging ), settings, metricsListener, logging, clock, ownsEventLoopGroup );
     }
 
@@ -87,7 +91,7 @@ public CompletionStage<Connection> acquire( BoltServerAddress address )
         log.trace( "Acquiring a connection from pool towards %s", address );
 
         assertNotClosed();
-        ChannelPool pool = getOrCreatePool( address );
+        ExtendedChannelPool pool = getOrCreatePool( address );
 
         ListenerEvent acquireEvent = metricsListener.createListenerEvent();
         metricsListener.beforeAcquiringOrCreating( address, acquireEvent );
@@ -97,7 +101,7 @@ public CompletionStage<Connection> acquire( BoltServerAddress address )
         {
             try
             {
-                processAcquisitionError( address, error );
+                processAcquisitionError( pool, address, error );
                 assertNotClosed( address, channel, pool );
                 Connection connection = new DirectConnection( channel, pool, clock, metricsListener );
 
@@ -156,7 +160,7 @@ public CompletionStage<Void> close()
             try
             {
                 nettyChannelTracker.prepareToCloseChannels();
-                for ( Map.Entry<BoltServerAddress,ChannelPool> entry : pools.entrySet() )
+                for ( Map.Entry<BoltServerAddress,ExtendedChannelPool> entry : pools.entrySet() )
                 {
                     BoltServerAddress address = entry.getKey();
                     ChannelPool pool = entry.getValue();
@@ -192,9 +196,9 @@ public boolean isOpen( BoltServerAddress address )
         return pools.containsKey( address );
     }
 
-    private ChannelPool getOrCreatePool( BoltServerAddress address )
+    private ExtendedChannelPool getOrCreatePool( BoltServerAddress address )
     {
-        ChannelPool pool = pools.get( address );
+        ExtendedChannelPool pool = pools.get( address );
         if ( pool != null )
         {
             return pool;
@@ -215,7 +219,7 @@ private ChannelPool getOrCreatePool( BoltServerAddress address )
         return pool;
     }
 
-    ChannelPool newPool( BoltServerAddress address )
+    ExtendedChannelPool newPool( BoltServerAddress address )
     {
         return new NettyChannelPool( address, connector, bootstrap, nettyChannelTracker, channelHealthChecker,
                 settings.connectionAcquisitionTimeout(), settings.maxConnectionPoolSize() );
@@ -226,7 +230,7 @@ private EventLoopGroup eventLoopGroup()
         return bootstrap.config().group();
     }
 
-    private void processAcquisitionError( BoltServerAddress serverAddress, Throwable error )
+    private void processAcquisitionError( ExtendedChannelPool pool, BoltServerAddress serverAddress, Throwable error )
     {
         Throwable cause = Futures.completionExceptionCause( error );
         if ( cause != null )
@@ -240,6 +244,13 @@ private void processAcquisitionError( BoltServerAddress serverAddress, Throwable
                         "Unable to acquire connection from the pool within configured maximum time of " +
                         settings.connectionAcquisitionTimeout() + "ms" );
             }
+            else if ( pool.isClosed() )
+            {
+                // There is a race condition where a thread tries to acquire a connection while the pool is closed by another concurrent thread.
+                // Treat as failed to obtain connection for a direct driver. For a routing driver, this error should be retried.
+                throw new ServiceUnavailableException( format( "Connection pool for server %s is closed while acquiring a connection.", serverAddress ),
+                        cause );
+            }
             else
             {
                 // some unknown error happened during connection acquisition, propagate it
@@ -272,4 +283,10 @@ public String toString()
     {
         return "ConnectionPoolImpl{" + "pools=" + pools + '}';
     }
+
+    // for testing only
+    ExtendedChannelPool getPool( BoltServerAddress address )
+    {
+        return pools.get( address );
+    }
 }

driver/src/main/java/org/neo4j/driver/internal/async/pool/ExtendedChannelPool.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2002-2019 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed 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.neo4j.driver.internal.async.pool;
+
+import io.netty.channel.pool.ChannelPool;
+
+public interface ExtendedChannelPool extends ChannelPool
+{
+    boolean isClosed();
+}