Fix file desc. leak with wrong credentials connect

Fixes #132.

Author: Totktonada

src/main/java/org/tarantool/TarantoolBase.java

@@ -40,11 +40,13 @@ public TarantoolBase() {
     public TarantoolBase(String username, String password, Socket socket) {
         super();
         try {
-            this.is = new DataInputStream(cis = new CountInputStreamImpl(socket.getInputStream()));
+            cis = new CountInputStreamImpl(socket.getInputStream())
+            is = new DataInputStream(cis);
             byte[] bytes = new byte[64];
             is.readFully(bytes);
             String firstLine = new String(bytes);
             if (!firstLine.startsWith(WELCOME)) {
+                closeStreams();
                 close();
                 throw new CommunicationException("Welcome message should starts with tarantool but starts with '" + firstLine + "'", new IllegalStateException("Invalid welcome packet"));
             }
@@ -56,23 +58,15 @@ public TarantoolBase(String username, String password, Socket socket) {
                 OutputStream os = socket.getOutputStream();
                 os.write(authPacket.array(), 0, authPacket.remaining());
                 os.flush();
-                readPacket(is);
+                readPacket();
                 Long code = (Long) headers.get(Key.CODE.getId());
                 if (code != 0) {
+                    closeStreams();
                     throw serverError(code, body.get(Key.ERROR.getId()));
                 }
             }
         } catch (IOException e) {
-            try {
-                is.close();
-            } catch (IOException ignored) {
-
-            }
-            try {
-                cis.close();
-            } catch (IOException ignored) {
-
-            }
+            closeStreams();
             throw new CommunicationException("Couldn't connect to tarantool", e);
         }
     }
@@ -130,7 +124,7 @@ protected ByteBuffer createPacket(Code code, Long syncId, Long schemaId, Object.
         return buffer;
     }
 
-    protected void readPacket(DataInputStream is) throws IOException {
+    protected void readPacket() throws IOException {
         int size = ((Number) msgPackLite.unpack(is)).intValue();
         long mark = cis.getBytesRead();
         headers = (Map<Integer, Object>) msgPackLite.unpack(is);
@@ -185,7 +179,6 @@ protected List<Map<String, Object>> readSqlResult(List<List<?>> data) {
         return values;
     }
 
-
     protected Long getSqlRowCount() {
         Map<Key, Object> info = (Map<Key, Object>) body.get(Key.SQL_INFO.getId());
         Number rowCount;
@@ -220,6 +213,21 @@ protected void closeChannel(SocketChannel channel) {
         }
     }
 
+    protected void closeStreams() {
+        if (is != null) {
+            try {
+                is.close();
+            } catch (IOException ignored) {
+            }
+        }
+        if (cis != null) {
+            try {
+                cis.close();
+            } catch (IOException ignored) {
+            }
+        }
+    }
+
     protected void validateArgs(Object[] args) {
         if (args != null) {
             for (int i = 0; i < args.length; i += 2) {

src/main/java/org/tarantool/TarantoolClientImpl.java

@@ -132,40 +132,34 @@ protected void reconnect(int retry, Throwable lastError) {
     }
 
     protected void connect(final SocketChannel channel) throws Exception {
+        closeStreams();
         try {
-            DataInputStream is = new DataInputStream(cis = new ByteBufferInputStream(channel));
+            cis = new ByteBufferInputStream(channel);
+            is = new DataInputStream(cis);
             byte[] bytes = new byte[64];
             is.readFully(bytes);
             String firstLine = new String(bytes);
             if (!firstLine.startsWith("Tarantool")) {
                 CommunicationException e = new CommunicationException("Welcome message should starts with tarantool " +
                         "but starts with '" + firstLine + "'", new IllegalStateException("Invalid welcome packet"));
 
+                closeStreams();
                 close(e);
                 throw e;
             }
             is.readFully(bytes);
             this.salt = new String(bytes);
             if (config.username != null && config.password != null) {
                 writeFully(channel, createAuthPacket(config.username, config.password));
-                readPacket(is);
+                readPacket();
                 Long code = (Long) headers.get(Key.CODE.getId());
                 if (code != 0) {
+                    closeStreams();
                     throw serverError(code, body.get(Key.ERROR.getId()));
                 }
             }
-            this.is = is;
         } catch (IOException e) {
-            try {
-                is.close();
-            } catch (IOException ignored) {
-
-            }
-            try {
-                cis.close();
-            } catch (IOException ignored) {
-
-            }
+            closeStreams();
             throw new CommunicationException("Couldn't connect to tarantool", e);
         }
         channel.configureBlocking(false);
@@ -358,7 +352,7 @@ protected void readThread() {
             while (!Thread.currentThread().isInterrupted()) {
                 try {
                     long code;
-                    readPacket(is);
+                    readPacket();
                     code = (Long) headers.get(Key.CODE.getId());
                     Long syncId = (Long) headers.get(Key.SYNC.getId());
                     CompletableFuture<?> future = futures.remove(syncId);

src/main/java/org/tarantool/TarantoolConnection.java

@@ -28,7 +28,7 @@ protected List<?> exec(Code code, Object... args) {
             ByteBuffer packet = createPacket(code, syncId.incrementAndGet(), null, args);
             out.write(packet.array(), 0, packet.remaining());
             out.flush();
-            readPacket(is);
+            readPacket();
             Long c = (Long) headers.get(Key.CODE.getId());
             if (c == 0) {
                 return (List) body.get(Key.DATA.getId());

src/test/java/org/tarantool/ClientReconnectIT.java

@@ -28,6 +28,7 @@
 public class ClientReconnectIT extends AbstractTarantoolConnectorIT {
     private static final String INSTANCE_NAME = "jdk-testing";
     private TarantoolClient client;
+    private static final int NON_EXIST_PORT = 3333;
 
     @AfterEach
     public void tearDown() {
@@ -302,4 +303,49 @@ public void run() {
 
         assertTrue(cnt.get() > threads.length);
     }
+
+    /**
+     * Verify that we don't exhaust a file descriptor limit (and so likely
+     * don't leak file descriptors) when trying to connect to a non existing
+     * node.
+     */
+    @Test
+    public void testReconnectNonExist() throws Exception {
+        SocketChannelProvider provider = new TestSocketChannelProvider(host,
+            NON_EXIST_PORT, RESTART_TIMEOUT);
+        TarantoolClientConfig config = makeClientConfig();
+        config.initTimeoutMillis = 100;
+        for (int i = 0; i < 100; ++i) {
+            try {
+                TarantoolClientImpl client = new TarantoolClientImpl(
+                    provider, config);
+            } catch (Exception ignored) {
+            }
+            if (i % 10 == 0)
+                System.out.println(Integer.toString(100 - i) +
+                                   " iterations remain");
+        }
+    }
+
+    /**
+     * Verify that we don't exhaust a file descriptor limit (and so likely
+     * don't leak file descriptors) when trying to connect to an existing node
+     * with wrong authentification credentials.
+     */
+    @Test
+    public void testReconnectWrongAuth() throws Exception {
+        TarantoolClientConfig config = makeClientConfig();
+        config.initTimeoutMillis = 100;
+        config.password = config.password + 'x';
+        for (int i = 0; i < 100; ++i) {
+            try {
+                TarantoolClientImpl client = new TarantoolClientImpl(
+                    socketChannelProvider, config);
+            } catch (Exception ignored) {
+            }
+            if (i % 10 == 0)
+                System.out.println(Integer.toString(100 - i) +
+                                   " iterations remain");
+        }
+    }
 }