Support auto refresh a list of cluster nodes

Refactor SocketChannelProvider implementations. Now we have two
SingleSocketChannelProviderImpl and RoundRobinSocketProviderImpl used by
simple and cluster clients respectively. To achieve this a
BaseSocketChannelProvider was extracted.

Add a service discovery implementation based on a tarantool stored
procedure which is called to obtain a new list of cluster nodes.

Integrate service discovery and current cluster client. The client now
is aware of potential nodes changing using a configurable background job
which periodically checks whether node addresses have changed. If so
the client refreshes the RoundRobinSocketProviderImpl and replaces old
nodes by new ones. It also requires some additional effort in case of
missing the current node in the list. We need to reconnect to another
node as soon as possible with a minimal delay between client requests.
To achieve this we currently try to catch a moment when the requests in
progress have been accomplished and we can finish reconnection process.

Closes: #34

Author: nicktorwald

src/main/java/org/tarantool/BaseSocketChannelProvider.java

@@ -0,0 +1,178 @@
+package org.tarantool;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.channels.SocketChannel;
+
+public abstract class BaseSocketChannelProvider implements SocketChannelProvider {
+
+    /**
+     * Special exception used to provide some sort of lazy-initialization
+     * feature for its derived implementations
+     *
+     * @see #getAddress(int, Throwable)
+     */
+    private final static CommunicationException NO_ADDRESS_AVAILABLE_EXCEPTION
+            = new CommunicationException("No addresses are available");
+
+    /**
+     * Limit of retries.
+     */
+    private int retriesLimit = RETRY_NO_LIMIT;
+
+    /**
+     * Timeout to establish socket connection with an individual server.
+     */
+    private int timeout = NO_TIMEOUT;
+
+    /**
+     * Tries to establish a new connection to the Tarantool instances.
+     *
+     * @param retryNumber number of current retry. Reset after successful connect.
+     * @param lastError   the last error occurs when reconnecting
+     *
+     * @return connected socket channel
+     *
+     * @throws CommunicationException if any I/O errors happen or there are
+     *                                no addresses available
+     */
+    @Override
+    public final SocketChannel get(int retryNumber, Throwable lastError) {
+        if (areRetriesExhausted(retryNumber)) {
+            throw new CommunicationException("Connection retries exceeded.", lastError);
+        }
+
+        long deadline = System.currentTimeMillis() + timeout;
+        while (!Thread.currentThread().isInterrupted()) {
+            try {
+                InetSocketAddress address = getAddress(retryNumber, lastError);
+                if (address == null) {
+                    checkTimeout(deadline, NO_ADDRESS_AVAILABLE_EXCEPTION);
+                } else {
+                    return openChannel(address);
+                }
+            } catch (IOException e) {
+                checkTimeout(deadline, e);
+            }
+        }
+        throw new CommunicationException("Thread interrupted.", new InterruptedException());
+    }
+
+    private void checkTimeout(long deadline, Exception e) {
+        long timeLeft = deadline - System.currentTimeMillis();
+        if (timeLeft <= 0) {
+            throw new CommunicationException("Connection time out.", e);
+        }
+        try {
+            Thread.sleep(timeLeft / 10);
+        } catch (InterruptedException ignored) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
+    /**
+     * Gets address to be used to establish a new connection
+     * Address can be null
+     *
+     * @param retryNumber reconnection attempt number
+     * @param lastError   reconnection reason
+     *
+     * @return available address which is depended on implementation
+     *
+     * @throws IOException if any I/O errors occur
+     */
+    protected abstract InetSocketAddress getAddress(int retryNumber, Throwable lastError) throws IOException;
+
+    /**
+     * Sets maximum amount of reconnect attempts to be made before an exception is raised.
+     * The retry count is maintained by a {@link #get(int, Throwable)} caller
+     * when a socket level connection was established.
+     * <p>
+     * Negative value means unlimited attempts.
+     *
+     * @param retriesLimit Limit of retries to use.
+     */
+    public void setRetriesLimit(int retriesLimit) {
+        this.retriesLimit = retriesLimit;
+    }
+
+    /**
+     * @return Maximum reconnect attempts to make before raising exception.
+     */
+    public int getRetriesLimit() {
+        return retriesLimit;
+    }
+
+    /**
+     * Parse a string address in the form of host[:port]
+     * and builds a socket address.
+     *
+     * @param address Server address.
+     *
+     * @return Socket address.
+     */
+    protected InetSocketAddress parseAddress(String address) {
+        int separatorPosition = address.indexOf(':');
+        String host = (separatorPosition < 0) ? address : address.substring(0, separatorPosition);
+        int port = (separatorPosition < 0) ? 3301 : Integer.parseInt(address.substring(separatorPosition + 1));
+        return new InetSocketAddress(host, port);
+    }
+
+    protected SocketChannel openChannel(InetSocketAddress socketAddress) throws IOException {
+        SocketChannel channel = null;
+        try {
+            channel = SocketChannel.open();
+            channel.socket().connect(socketAddress, timeout);
+            return channel;
+        } catch (IOException e) {
+            if (channel != null) {
+                try {
+                    channel.close();
+                } catch (IOException ignored) {
+                    // No-op.
+                }
+            }
+            throw e;
+        }
+    }
+
+    /**
+     * Sets maximum amount of time to wait for a socket connection establishment
+     * with an individual server.
+     * <p>
+     * Zero means infinite timeout.
+     *
+     * @param timeout timeout value, ms.
+     *
+     * @throws IllegalArgumentException if timeout is negative.
+     */
+    public void setTimeout(int timeout) {
+        if (timeout < 0) {
+            throw new IllegalArgumentException("timeout is negative.");
+        }
+        this.timeout = timeout;
+    }
+
+    /**
+     * @return Maximum amount of time to wait for a socket connection establishment
+     *         with an individual server.
+     */
+    public int getTimeout() {
+        return timeout;
+    }
+
+    /**
+     * Provides a decision on whether retries limit is hit.
+     *
+     * @param retries Current count of retries.
+     *
+     * @return {@code true} if retries are exhausted.
+     */
+    private boolean areRetriesExhausted(int retries) {
+        int limit = getRetriesLimit();
+        if (limit < 0) {
+            return false;
+        }
+        return retries >= limit;
+    }
+}

src/main/java/org/tarantool/RefreshableSocketProvider.java

@@ -0,0 +1,12 @@
+package org.tarantool;
+
+import java.net.SocketAddress;
+import java.util.Collection;
+
+public interface RefreshableSocketProvider {
+
+    Collection<SocketAddress> getAddresses();
+
+    void refreshAddresses(Collection<String> addresses);
+
+}

src/main/java/org/tarantool/RoundRobinSocketProviderImpl.java

@@ -2,181 +2,155 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.nio.channels.SocketChannel;
+import java.net.SocketAddress;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
 
 /**
  * Basic reconnection strategy that changes addresses in a round-robin fashion.
  * To be used with {@link TarantoolClientImpl}.
  */
-public class RoundRobinSocketProviderImpl implements SocketChannelProvider {
-    /** Timeout to establish socket connection with an individual server. */
-    private int timeout; // 0 is infinite.
-    /** Limit of retries. */
-    private int retriesLimit = -1; // No-limit.
-    /** Server addresses as configured. */
-    private final String[] addrs;
-    /** Socket addresses. */
-    private final InetSocketAddress[] sockAddrs;
-    /** Current position within {@link #sockAddrs} array. */
-    private int pos;
+public class RoundRobinSocketProviderImpl extends BaseSocketChannelProvider implements RefreshableSocketProvider {
+
+    private static final int UNSET_POSITION = -1;
 
     /**
-     * Constructs an instance.
-     *
-     * @param addrs Array of addresses in a form of [host]:[port].
+     * Socket addresses pool.
      */
-    public RoundRobinSocketProviderImpl(String... addrs) {
-        if (addrs == null || addrs.length == 0)
-            throw new IllegalArgumentException("addrs is null or empty.");
-
-        this.addrs = Arrays.copyOf(addrs, addrs.length);
-
-        sockAddrs = new InetSocketAddress[this.addrs.length];
-
-        for (int i = 0; i < this.addrs.length; i++) {
-            sockAddrs[i] = parseAddress(this.addrs[i]);
-        }
-    }
+    private final List<InetSocketAddress> socketAddresses = new ArrayList<>();
 
     /**
-     * @return Configured addresses in a form of [host]:[port].
+     * Current position within {@link #socketAddresses} list.
+     * <p>
+     * It is {@link #UNSET_POSITION} when no addresses from
+     * the {@link #socketAddresses} pool have been processed yet.
+     * <p>
+     * When this provider receives new addresses it tries
+     * to look for a new position for the last used address or
+     * sets the position to {@link #UNSET_POSITION} otherwise.
+     *
+     * @see #getLastObtainedAddress()
+     * @see #refreshAddresses(Collection)
      */
-    public String[] getAddresses() {
-        return this.addrs;
-    }
+    private AtomicInteger currentPosition = new AtomicInteger(UNSET_POSITION);
 
     /**
-     * Sets maximum amount of time to wait for a socket connection establishment
-     * with an individual server.
+     * Address list lock for a thread-safe access to it
+     * when a refresh operation occurs
      *
-     * Zero means infinite timeout.
-     *
-     * @param timeout Timeout value, ms.
-     * @return {@code this}.
-     * @throws IllegalArgumentException If timeout is negative.
+     * @see RefreshableSocketProvider#refreshAddresses(Collection)
      */
-    public RoundRobinSocketProviderImpl setTimeout(int timeout) {
-        if (timeout < 0)
-            throw new IllegalArgumentException("timeout is negative.");
-
-        this.timeout = timeout;
-
-        return this;
-    }
+    private ReadWriteLock addressListLock = new ReentrantReadWriteLock();
 
     /**
-     * @return Maximum amount of time to wait for a socket connection establishment
-     *         with an individual server.
+     * Constructs an instance.
+     *
+     * @param addresses optional array of addresses in a form of host[:port]
+     * @throws IllegalArgumentException if addresses aren't provided
      */
-    public int getTimeout() {
-        return timeout;
+    public RoundRobinSocketProviderImpl(String... addresses) {
+        updateAddressList(Arrays.asList(addresses));
+    }
+
+    private void updateAddressList(Collection<String> addresses) {
+        if (addresses == null || addresses.isEmpty()) {
+            throw new IllegalArgumentException("At least one address must be provided");
+        }
+        Lock writeLock = addressListLock.writeLock();
+        writeLock.lock();
+        try {
+            InetSocketAddress lastAddress = getLastObtainedAddress();
+            socketAddresses.clear();
+            addresses.stream()
+                    .map(this::parseAddress)
+                    .collect(Collectors.toCollection(() -> socketAddresses));
+            if (lastAddress != null) {
+                int recoveredPosition = socketAddresses.indexOf(lastAddress);
+                currentPosition.set(recoveredPosition);
+            } else {
+                currentPosition.set(UNSET_POSITION);
+            }
+        } finally {
+            writeLock.unlock();
+        }
     }
 
     /**
-     * Sets maximum amount of reconnect attempts to be made before an exception is raised.
-     * The retry count is maintained by a {@link #get(int, Throwable)} caller
-     * when a socket level connection was established.
-     *
-     * Negative value means unlimited.
-     *
-     * @param retriesLimit Limit of retries to use.
-     * @return {@code this}.
+     * @return resolved socket addresses
      */
-    public RoundRobinSocketProviderImpl setRetriesLimit(int retriesLimit) {
-        this.retriesLimit = retriesLimit;
-
-        return this;
+    public List<SocketAddress> getAddresses() {
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            return Collections.unmodifiableList(this.socketAddresses);
+        } finally {
+            readLock.unlock();
+        }
     }
 
     /**
-     * @return Maximum reconnect attempts to make before raising exception.
+     * Gets last used address from the pool if it exists
+     *
+     * @return last obtained address or <code>null</code>
+     *         if {@link #currentPosition} has {@link #UNSET_POSITION} value
      */
-    public int getRetriesLimit() {
-        return retriesLimit;
+    protected InetSocketAddress getLastObtainedAddress() {
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            int index = currentPosition.get();
+            return index != UNSET_POSITION ? socketAddresses.get(index) : null;
+        } finally {
+            readLock.unlock();
+        }
     }
 
-    /** {@inheritDoc} */
     @Override
-    public SocketChannel get(int retryNumber, Throwable lastError) {
-        if (areRetriesExhausted(retryNumber)) {
-            throw new CommunicationException("Connection retries exceeded.", lastError);
-        }
-        int attempts = getAddressCount();
-        long deadline = System.currentTimeMillis() + timeout * attempts;
-        while (!Thread.currentThread().isInterrupted()) {
-            SocketChannel channel = null;
-            try {
-                channel = SocketChannel.open();
-                InetSocketAddress addr = getNextSocketAddress();
-                channel.socket().connect(addr, timeout);
-                return channel;
-            } catch (IOException e) {
-                if (channel != null) {
-                    try {
-                        channel.close();
-                    } catch (IOException ignored) {
-                        // No-op.
-                    }
-                }
-                long now = System.currentTimeMillis();
-                if (deadline <= now) {
-                    throw new CommunicationException("Connection time out.", e);
-                }
-                if (--attempts == 0) {
-                    // Tried all addresses without any lack, but still have time.
-                    attempts = getAddressCount();
-                    try {
-                        Thread.sleep((deadline - now) / attempts);
-                    } catch (InterruptedException ignored) {
-                        Thread.currentThread().interrupt();
-                    }
-                }
-            }
-        }
-        throw new CommunicationException("Thread interrupted.", new InterruptedException());
+    protected InetSocketAddress getAddress(int retryNumber, Throwable lastError) throws IOException {
+        return getNextSocketAddress();
     }
 
     /**
      * @return Number of configured addresses.
      */
     protected int getAddressCount() {
-        return sockAddrs.length;
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            return socketAddresses.size();
+        } finally {
+            readLock.unlock();
+        }
     }
 
     /**
-     * @return Socket address to use for the next reconnection attempt.
+     * @return Socket address to use for the next reconnection attempt
      */
     protected InetSocketAddress getNextSocketAddress() {
-        InetSocketAddress res = sockAddrs[pos];
-        pos = (pos + 1) % sockAddrs.length;
-        return res;
-    }
-
-    /**
-     * Parse a string address in the form of [host]:[port]
-     * and builds a socket address.
-     *
-     * @param addr Server address.
-     * @return Socket address.
-     */
-    protected InetSocketAddress parseAddress(String addr) {
-        int idx = addr.indexOf(':');
-        String host = (idx < 0) ? addr : addr.substring(0, idx);
-        int port = (idx < 0) ? 3301 : Integer.parseInt(addr.substring(idx + 1));
-        return new InetSocketAddress(host, port);
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            int position = currentPosition.updateAndGet(i -> (i + 1) % socketAddresses.size());
+            return socketAddresses.get(position);
+        } finally {
+            readLock.unlock();
+        }
     }
 
     /**
-     * Provides a decision on whether retries limit is hit.
+     * @param addresses list of addresses to be applied
      *
-     * @param retries Current count of retries.
-     * @return {@code true} if retries are exhausted.
+     * @throws IllegalArgumentException if addresses list is empty
      */
-    private boolean areRetriesExhausted(int retries) {
-        int limit = getRetriesLimit();
-        if (limit < 0)
-            return false;
-        return retries >= limit;
+    public void refreshAddresses(Collection<String> addresses) {
+        updateAddressList(addresses);
     }
 }

src/main/java/org/tarantool/SingleSocketChannelProviderImpl.java

@@ -0,0 +1,43 @@
+package org.tarantool;
+
+import org.tarantool.util.StringUtils;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+
+/**
+ * Simple provider that produces a single connection
+ * To be used with {@link TarantoolClientImpl}.
+ */
+public class SingleSocketChannelProviderImpl extends BaseSocketChannelProvider {
+
+    private InetSocketAddress address;
+
+    /**
+     *  Creates a simple provider
+     *
+     * @param address instance address
+     */
+    public SingleSocketChannelProviderImpl(String address) {
+        setAddress(address);
+    }
+
+    public SocketAddress getAddress() {
+        return address;
+    }
+
+    public void setAddress(String address) {
+        if (StringUtils.isBlank(address)) {
+            throw new IllegalArgumentException("address must not be empty");
+        }
+
+        this.address = parseAddress(address);
+    }
+
+    @Override
+    protected InetSocketAddress getAddress(int retryNumber, Throwable lastError) throws IOException {
+        return address;
+    }
+
+}

src/main/java/org/tarantool/SocketChannelProvider.java

@@ -4,6 +4,10 @@
 import java.nio.channels.SocketChannel;
 
 public interface SocketChannelProvider {
+
+    int RETRY_NO_LIMIT = -1;
+    int NO_TIMEOUT = 0;
+
     /**
      * Provides socket channel to init restore connection.
      * You could change hosts on fail and sleep between retries in this method

src/main/java/org/tarantool/TarantoolClientConfig.java

@@ -1,53 +1,50 @@
 package org.tarantool;
 
-
-import java.util.concurrent.TimeUnit;
-
 public class TarantoolClientConfig {
 
     /**
-     * username and password for authorization
+     * Auth-related data
      */
     public String username;
-
     public String password;
 
     /**
-     * default ByteArrayOutputStream size  when make query serialization
+     * Default request size when make query serialization
      */
     public int defaultRequestSize = 4096;
 
     /**
-     * initial size for map which holds futures of sent request
+     * Initial capacity for the map which holds futures of sent request
      */
     public int predictedFutures = (int) ((1024 * 1024) / 0.75) + 1;
 
-
     public int writerThreadPriority = Thread.NORM_PRIORITY;
-
     public int readerThreadPriority = Thread.NORM_PRIORITY;
 
-
     /**
-     * shared buffer is place where client collect requests when socket is busy on write
+     *  Shared buffer size (place where client collects requests
+     *  when socket is busy on write)
      */
     public int sharedBufferSize = 8 * 1024 * 1024;
+
     /**
-     * not put request into the shared buffer if request size is ge directWriteFactor * sharedBufferSize
+     * Factor to calculate a threshold whether request will be accommodated
+     * in the shared buffer.
+     * if request size exceeds <code>directWriteFactor * sharedBufferSize</code>
+     * request is sent directly.
      */
     public double directWriteFactor = 0.5d;
 
     /**
-     *  Use old call command https://github.com/tarantool/doc/issues/54,
-     *  please ensure that you server supports new call command
+     * Use old call command https://github.com/tarantool/doc/issues/54,
+     * please ensure that you server supports new call command
      */
     public boolean useNewCall = false;
 
     /**
-     * Any blocking ops timeout
+     * Limits for synchronous operations
      */
-    public long initTimeoutMillis = 60*1000L;
-
-    public long writeTimeoutMillis = 60*1000L;
+    public long initTimeoutMillis = 60 * 1000L;
+    public long writeTimeoutMillis = 60 * 1000L;
 
 }

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

@@ -22,33 +22,35 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.LockSupport;
 import java.util.concurrent.locks.ReentrantLock;
 
 
 public class TarantoolClientImpl extends TarantoolBase<Future<?>> implements TarantoolClient {
+
     public static final CommunicationException NOT_INIT_EXCEPTION = new CommunicationException("Not connected, initializing connection");
+
     protected TarantoolClientConfig config;
 
     /**
      * External
      */
     protected SocketChannelProvider socketProvider;
+    protected SocketChannel channel;
+    protected ReadableViaSelectorChannel readChannel;
+
     protected volatile Exception thumbstone;
 
     protected Map<Long, TarantoolOp<?>> futures;
-    protected AtomicInteger wait = new AtomicInteger();
+    protected AtomicInteger pendingResponsesCount = new AtomicInteger();
     /**
      * Write properties
      */
-    protected SocketChannel channel;
-    protected ReadableViaSelectorChannel readChannel;
-
     protected ByteBuffer sharedBuffer;
-    protected ByteBuffer writerBuffer;
     protected ReentrantLock bufferLock = new ReentrantLock(false);
     protected Condition bufferNotEmpty = bufferLock.newCondition();
     protected Condition bufferEmpty = bufferLock.newCondition();
+
+    protected ByteBuffer writerBuffer;
     protected ReentrantLock writeLock = new ReentrantLock(true);
 
     /**
@@ -66,18 +68,23 @@ public class TarantoolClientImpl extends TarantoolBase<Future<?>> implements Tar
     protected Thread reader;
     protected Thread writer;
 
+    protected final ReentrantLock connectorLock = new ReentrantLock();
+    protected final Condition reconnectRequired = connectorLock.newCondition();
+
     protected Thread connector = new Thread(new Runnable() {
         @Override
         public void run() {
             while (!Thread.currentThread().isInterrupted()) {
-                if (state.compareAndSet(StateHelper.RECONNECT, 0)) {
-                    reconnect(0, thumbstone);
-                }
-                LockSupport.park(state);
+                reconnect(0, thumbstone);
+                awaitReconnection();
             }
         }
     });
 
+    public TarantoolClientImpl(String address, TarantoolClientConfig config) {
+        this(new SingleSocketChannelProviderImpl(address), config);
+    }
+
     public TarantoolClientImpl(SocketChannelProvider socketProvider, TarantoolClientConfig config) {
         super();
         this.thumbstone = NOT_INIT_EXCEPTION;
@@ -99,6 +106,11 @@ public TarantoolClientImpl(SocketChannelProvider socketProvider, TarantoolClient
             this.fireAndForgetOps.setCallCode(Code.CALL);
             this.composableAsyncOps.setCallCode(Code.CALL);
         }
+
+        startConnector(config);
+    }
+
+    private void startConnector(TarantoolClientConfig config) {
         connector.start();
         try {
             if (!waitAlive(config.initTimeoutMillis, TimeUnit.MILLISECONDS)) {
@@ -130,25 +142,22 @@ protected void reconnect(int retry, Throwable lastError) {
             } catch (Exception e) {
                 closeChannel(channel);
                 lastError = e;
-                if (e instanceof InterruptedException)
+                if (e instanceof InterruptedException) {
                     Thread.currentThread().interrupt();
+                }
             }
         }
     }
 
     protected void connect(final SocketChannel channel) throws Exception {
         try {
-            TarantoolGreeting greeting = ProtoUtils.connect(channel,
-                config.username, config.password);
+            TarantoolGreeting greeting = ProtoUtils.connect(channel, config.username, config.password);
             this.serverVersion = greeting.getServerVersion();
         } catch (IOException e) {
-            try {
-                channel.close();
-            } catch (IOException ignored) {
-            }
-
+            closeChannel(channel);
             throw new CommunicationException("Couldn't connect to tarantool", e);
         }
+
         channel.configureBlocking(false);
         this.channel = channel;
         this.readChannel = new ReadableViaSelectorChannel(channel);
@@ -160,11 +169,21 @@ protected void connect(final SocketChannel channel) throws Exception {
             bufferLock.unlock();
         }
         this.thumbstone = null;
+        pendingResponsesCount.set(0);
         startThreads(channel.socket().getRemoteSocketAddress().toString());
     }
 
     protected void startThreads(String threadName) throws InterruptedException {
         final CountDownLatch init = new CountDownLatch(2);
+
+        if (reader != null) {
+            reader.join(config.initTimeoutMillis / 2);
+        }
+        if (writer != null) {
+            writer.join(config.initTimeoutMillis / 2);
+        }
+        state.release(StateHelper.RECONNECT);
+
         reader = new Thread(new Runnable() {
             @Override
             public void run() {
@@ -174,8 +193,9 @@ public void run() {
                         readThread();
                     } finally {
                         state.release(StateHelper.READING);
-                        if (state.compareAndSet(0, StateHelper.RECONNECT))
-                            LockSupport.unpark(connector);
+                        if (state.compareAndSet(StateHelper.UNINITIALIZED, StateHelper.RECONNECT)) {
+                            signalForReconnection();
+                        }
                     }
                 }
             }
@@ -189,8 +209,9 @@ public void run() {
                         writeThread();
                     } finally {
                         state.release(StateHelper.WRITING);
-                        if (state.compareAndSet(0, StateHelper.RECONNECT))
-                            LockSupport.unpark(connector);
+                        if (state.compareAndSet(StateHelper.UNINITIALIZED, StateHelper.RECONNECT)) {
+                            signalForReconnection();
+                        }
                     }
                 }
             }
@@ -300,7 +321,7 @@ protected void sharedWrite(ByteBuffer buffer) throws InterruptedException, Timeo
                     }
                 }
                 sharedBuffer.put(buffer);
-                wait.incrementAndGet();
+                pendingResponsesCount.incrementAndGet();
                 bufferNotEmpty.signalAll();
                 stats.buffered++;
             } finally {
@@ -323,7 +344,7 @@ private boolean directWrite(ByteBuffer buffer) throws InterruptedException, IOEx
                     }
                     writeFully(channel, buffer);
                     stats.directWrite++;
-                    wait.incrementAndGet();
+                    pendingResponsesCount.incrementAndGet();
                 } finally {
                     writeLock.unlock();
                 }
@@ -337,25 +358,21 @@ private boolean directWrite(ByteBuffer buffer) throws InterruptedException, IOEx
     }
 
     protected void readThread() {
-        try {
-            while (!Thread.currentThread().isInterrupted()) {
-                try {
-                    TarantoolPacket packet = ProtoUtils.readPacket(readChannel);
+        while (!Thread.currentThread().isInterrupted()) {
+            try {
+                TarantoolPacket packet = ProtoUtils.readPacket(readChannel);
 
-                    Map<Integer, Object> headers = packet.getHeaders();
+                Map<Integer, Object> headers = packet.getHeaders();
 
-                    Long syncId = (Long) headers.get(Key.SYNC.getId());
-                    TarantoolOp<?> future = futures.remove(syncId);
-                    stats.received++;
-                    wait.decrementAndGet();
-                    complete(packet, future);
-                } catch (Exception e) {
-                    die("Cant read answer", e);
-                    return;
-                }
+                Long syncId = (Long) headers.get(Key.SYNC.getId());
+                TarantoolOp<?> future = futures.remove(syncId);
+                stats.received++;
+                pendingResponsesCount.decrementAndGet();
+                complete(packet, future);
+            } catch (Exception e) {
+                die("Cant read answer", e);
+                return;
             }
-        } catch (Exception e) {
-            die("Cant init thread", e);
         }
     }
 
@@ -421,9 +438,9 @@ protected void completeSql(CompletableFuture<?> future, TarantoolPacket pack) {
         }
     }
 
-    protected <T> T syncGet(Future<T> r) {
+    protected <T> T syncGet(Future<T> result) {
         try {
-            return r.get();
+            return result.get();
         } catch (ExecutionException e) {
             if (e.getCause() instanceof CommunicationException) {
                 throw (CommunicationException) e.getCause();
@@ -454,7 +471,6 @@ public void close() {
     protected void close(Exception e) {
         if (state.close()) {
             connector.interrupt();
-
             die(e.getMessage(), e);
         }
     }
@@ -468,14 +484,44 @@ protected void stopIO() {
         }
         if (readChannel != null) {
             try {
-                readChannel.close();//also closes this.channel
+                readChannel.close(); // also closes this.channel
             } catch (IOException ignored) {
 
             }
         }
         closeChannel(channel);
     }
 
+    /**
+     * Blocks until a reconnection process can be carried on
+     * @see #signalForReconnection()
+     */
+    private void awaitReconnection() {
+        connectorLock.lock();
+        try {
+            while (state.getState() != StateHelper.RECONNECT) {
+                reconnectRequired.await();
+            }
+        } catch (InterruptedException ignored) {
+            Thread.currentThread().interrupt();
+        } finally {
+            connectorLock.unlock();
+        }
+    }
+
+    /**
+     * Signals to the connector that reconnection process can be performed
+     * @see #awaitReconnection()
+     */
+    private void signalForReconnection() {
+        connectorLock.lock();
+        try {
+            reconnectRequired.signal();
+        } finally {
+            connectorLock.unlock();
+        }
+    }
+
     @Override
     public boolean isAlive() {
         return state.getState() == StateHelper.ALIVE && thumbstone == null;
@@ -498,7 +544,7 @@ public TarantoolClientOps<Integer, List<?>, Object, List<?>> syncOps() {
 
     @Override
     public TarantoolClientOps<Integer, List<?>, Object, Future<List<?>>> asyncOps() {
-        return (TarantoolClientOps)this;
+        return (TarantoolClientOps) this;
     }
 
     @Override
@@ -514,7 +560,7 @@ public TarantoolClientOps<Integer, List<?>, Object, Long> fireAndForgetOps() {
 
     @Override
     public TarantoolSQLOps<Object, Long, List<Map<String, Object>>> sqlSyncOps() {
-        return new TarantoolSQLOps<Object, Long, List<Map<String,Object>>>() {
+        return new TarantoolSQLOps<Object, Long, List<Map<String, Object>>>() {
 
             @Override
             public Long update(String sql, Object... bind) {
@@ -530,7 +576,7 @@ public List<Map<String, Object>> query(String sql, Object... bind) {
 
     @Override
     public TarantoolSQLOps<Object, Future<Long>, Future<List<Map<String, Object>>>> sqlAsyncOps() {
-        return new TarantoolSQLOps<Object, Future<Long>, Future<List<Map<String,Object>>>>() {
+        return new TarantoolSQLOps<Object, Future<Long>, Future<List<Map<String, Object>>>>() {
             @Override
             public Future<Long> update(String sql, Object... bind) {
                 return (Future<Long>) exec(Code.EXECUTE, Key.SQL_TEXT, sql, Key.SQL_BIND, bind);
@@ -591,7 +637,7 @@ public void close() {
     }
 
     protected boolean isDead(CompletableFuture<?> q) {
-        if (TarantoolClientImpl.this.thumbstone != null) {
+        if (this.thumbstone != null) {
             fail(q, new CommunicationException("Connection is dead", thumbstone));
             return true;
         }
@@ -618,6 +664,7 @@ public TarantoolClientStats getStats() {
      * Manages state changes.
      */
     protected final class StateHelper {
+        static final int UNINITIALIZED = 0;
         static final int READING = 1;
         static final int WRITING = 2;
         static final int ALIVE = READING | WRITING;
@@ -627,7 +674,7 @@ protected final class StateHelper {
         private final AtomicInteger state;
 
         private final AtomicReference<CountDownLatch> nextAliveLatch =
-            new AtomicReference<CountDownLatch>(new CountDownLatch(1));
+                new AtomicReference<>(new CountDownLatch(1));
 
         private final CountDownLatch closedLatch = new CountDownLatch(1);
 
@@ -640,7 +687,7 @@ protected int getState() {
         }
 
         protected boolean close() {
-            for (;;) {
+            for (; ; ) {
                 int st = getState();
                 if ((st & CLOSED) == CLOSED)
                     return false;
@@ -650,24 +697,29 @@ protected boolean close() {
         }
 
         protected boolean acquire(int mask) {
-            for (;;) {
-                int st = getState();
-                if ((st & CLOSED) == CLOSED)
+            for (; ; ) {
+                int currentState = getState();
+                if ((currentState & CLOSED) == CLOSED) {
                     return false;
-
-                if ((st & mask) != 0)
+                }
+                if ((currentState & RECONNECT) > mask) {
+                    return false;
+                }
+                if ((currentState & mask) != 0) {
                     throw new IllegalStateException("State is already " + mask);
-
-                if (compareAndSet(st, st | mask))
+                }
+                if (compareAndSet(currentState, currentState | mask)) {
                     return true;
+                }
             }
         }
 
         protected void release(int mask) {
-            for (;;) {
+            for (; ; ) {
                 int st = getState();
-                if (compareAndSet(st, st & ~mask))
+                if (compareAndSet(st, st & ~mask)) {
                     return;
+                }
             }
         }
 
@@ -709,7 +761,7 @@ private CountDownLatch getStateLatch(int state) {
                 CountDownLatch latch = nextAliveLatch.get();
                 /* It may happen so that an error is detected but the state is still alive.
                  Wait for the 'next' alive state in such cases. */
-                return  (getState() == ALIVE && thumbstone == null) ? null : latch;
+                return (getState() == ALIVE && thumbstone == null) ? null : latch;
             }
             return null;
         }

src/main/java/org/tarantool/TarantoolClusterClient.java

@@ -1,13 +1,23 @@
 package org.tarantool;
 
+import org.tarantool.cluster.TarantoolClusterDiscoverer;
+import org.tarantool.cluster.TarantoolClusterStoredFunctionDiscoverer;
+import org.tarantool.protocol.TarantoolPacket;
+import org.tarantool.util.StringUtils;
+
+import java.io.IOException;
+import java.net.SocketAddress;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
-
-import static org.tarantool.TarantoolClientImpl.StateHelper.CLOSED;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.StampedLock;
 
 /**
  * Basic implementation of a client that may work with the cluster
@@ -17,18 +27,30 @@
  * unless the configured expiration time is over.
  */
 public class TarantoolClusterClient extends TarantoolClientImpl {
-    /* Need some execution context to retry writes. */
+
+    /**
+     * Need some execution context to retry writes.
+     */
     private Executor executor;
 
-    /* Collection of operations to be retried. */
-    private ConcurrentHashMap<Long, ExpirableOp<?>> retries = new ConcurrentHashMap<Long, ExpirableOp<?>>();
+    /**
+     * Discovery activity
+     */
+    private ScheduledExecutorService instancesDiscoveryExecutor;
+    private Runnable instancesDiscovererTask;
+    private StampedLock discoveryLock = new StampedLock();
 
     /**
-     * @param config Configuration.
-     * @param addrs  Array of addresses in the form of [host]:[port].
+     * Collection of operations to be retried.
      */
-    public TarantoolClusterClient(TarantoolClusterClientConfig config, String... addrs) {
-        this(config, new RoundRobinSocketProviderImpl(addrs).setTimeout(config.operationExpiryTimeMillis));
+    private ConcurrentHashMap<Long, ExpirableOp<?>> retries = new ConcurrentHashMap<>();
+
+    /**
+     * @param config    Configuration.
+     * @param addresses Array of addresses in the form of host[:port].
+     */
+    public TarantoolClusterClient(TarantoolClusterClientConfig config, String... addresses) {
+        this(config, makeClusterSocketProvider(addresses, config.operationExpiryTimeMillis));
     }
 
     /**
@@ -38,13 +60,32 @@ public TarantoolClusterClient(TarantoolClusterClientConfig config, String... add
     public TarantoolClusterClient(TarantoolClusterClientConfig config, SocketChannelProvider provider) {
         super(provider, config);
 
-        this.executor = config.executor == null ?
-                Executors.newSingleThreadExecutor() : config.executor;
+        this.executor = config.executor == null
+                ? Executors.newSingleThreadExecutor()
+                : config.executor;
+
+        if (StringUtils.isNotBlank(config.clusterDiscoveryEntryFunction)) {
+            this.instancesDiscovererTask =
+                    createDiscoveryTask(new TarantoolClusterStoredFunctionDiscoverer(config, this));
+            this.instancesDiscoveryExecutor
+                    = Executors.newSingleThreadScheduledExecutor(new TarantoolThreadDaemonFactory("tarantoolDiscoverer"));
+            int delay = config.clusterDiscoveryDelayMillis > 0
+                ? config.clusterDiscoveryDelayMillis
+                : TarantoolClusterClientConfig.DEFAULT_CLUSTER_DISCOVERY_DELAY_MILLIS;
+
+            // todo: it's better to start a job later (out of ctor)
+            this.instancesDiscoveryExecutor.scheduleWithFixedDelay(
+                    this.instancesDiscovererTask,
+                    0,
+                    delay,
+                    TimeUnit.MILLISECONDS
+            );
+        }
     }
 
     @Override
     protected boolean isDead(CompletableFuture<?> q) {
-        if ((state.getState() & CLOSED) != 0) {
+        if ((state.getState() & StateHelper.CLOSED) != 0) {
             q.completeExceptionally(new CommunicationException("Connection is dead", thumbstone));
             return true;
         }
@@ -61,21 +102,28 @@ protected CompletableFuture<?> doExec(Code code, Object[] args) {
         long sid = syncId.incrementAndGet();
         ExpirableOp<?> future = makeFuture(sid, code, args);
 
-        if (isDead(future)) {
-            return future;
-        }
-        futures.put(sid, future);
-        if (isDead(future)) {
-            futures.remove(sid);
-            return future;
-        }
+        long stamp = discoveryLock.readLock();
         try {
-            write(code, sid, null, args);
-        } catch (Exception e) {
-            futures.remove(sid);
-            fail(future, e);
+            if (isDead(future)) {
+                return future;
+            }
+            futures.put(sid, future);
+            if (isDead(future)) {
+                futures.remove(sid);
+                return future;
+            }
+
+            try {
+                write(code, sid, null, args);
+            } catch (Exception e) {
+                futures.remove(sid);
+                fail(future, e);
+            }
+
+            return future;
+        } finally {
+            discoveryLock.unlock(stamp);
         }
-        return future;
     }
 
     @Override
@@ -99,6 +147,10 @@ protected boolean checkFail(CompletableFuture<?> q, Exception e) {
     protected void close(Exception e) {
         super.close(e);
 
+        if (instancesDiscoveryExecutor != null) {
+            instancesDiscoveryExecutor.shutdownNow();
+        }
+
         if (retries == null) {
             // May happen within constructor.
             return;
@@ -133,27 +185,109 @@ protected void onReconnect() {
             // First call is before the constructor finished. Skip it.
             return;
         }
-        Collection<ExpirableOp<?>> futuresToRetry = new ArrayList<ExpirableOp<?>>(retries.values());
+        Collection<ExpirableOp<?>> futuresToRetry = new ArrayList<>(retries.values());
         retries.clear();
         long now = System.currentTimeMillis();
         for (final ExpirableOp<?> future : futuresToRetry) {
             if (!future.hasExpired(now)) {
-                executor.execute(new Runnable() {
-                    @Override
-                    public void run() {
-                        futures.put(future.getId(), future);
-                        try {
-                            write(future.getCode(), future.getId(), null, future.getArgs());
-                        } catch (Exception e) {
-                            futures.remove(future.getId());
-                            fail(future, e);
-                        }
-                    }
-                });
+                executor.execute(() -> resend(future));
             }
         }
     }
 
+    private void resend(ExpirableOp<?> future) {
+        futures.put(future.getId(), future);
+        try {
+            write(future.getCode(), future.getId(), null, future.getArgs());
+        } catch (Exception e) {
+            futures.remove(future.getId());
+            fail(future, e);
+        }
+    }
+
+    @Override
+    protected void complete(TarantoolPacket packet, TarantoolOp<?> future) {
+        super.complete(packet, future);
+        RefreshableSocketProvider provider = getRefreshableSocketProvider();
+        if (provider != null) {
+            renewConnectionIfRequired(provider.getAddresses());
+        }
+    }
+
+    protected void onInstancesRefreshed(Set<String> instances) {
+        RefreshableSocketProvider provider = getRefreshableSocketProvider();
+        if (provider != null) {
+            provider.refreshAddresses(instances);
+            renewConnectionIfRequired(provider.getAddresses());
+        }
+    }
+
+    private RefreshableSocketProvider getRefreshableSocketProvider() {
+        return socketProvider instanceof RefreshableSocketProvider
+                ? (RefreshableSocketProvider) socketProvider
+                : null;
+    }
+
+    private void renewConnectionIfRequired(Collection<SocketAddress> addresses) {
+        if (pendingResponsesCount.get() > 0 || !isAlive()) {
+            return;
+        }
+        SocketAddress addressInUse = getCurrentAddressOrNull();
+        if (!(addressInUse == null || addresses.contains(addressInUse))) {
+            long stamp = discoveryLock.tryWriteLock();
+            if (!discoveryLock.validate(stamp)) {
+                return;
+            }
+            try {
+                if (pendingResponsesCount.get() == 0) {
+                    stopIO();
+                }
+            } finally {
+                discoveryLock.unlock(stamp);
+            }
+        }
+    }
+
+    private SocketAddress getCurrentAddressOrNull() {
+        try {
+            return channel.getRemoteAddress();
+        } catch (IOException ignored) {
+            return null;
+        }
+    }
+
+    public void refreshInstances() {
+        if (instancesDiscovererTask != null) {
+            instancesDiscovererTask.run();
+        }
+    }
+
+    private static RoundRobinSocketProviderImpl makeClusterSocketProvider(String[] addresses,
+                                                                          int connectionTimeout) {
+        RoundRobinSocketProviderImpl socketProvider = new RoundRobinSocketProviderImpl(addresses);
+        socketProvider.setTimeout(connectionTimeout);
+        return socketProvider;
+    }
+
+    private Runnable createDiscoveryTask(TarantoolClusterDiscoverer serviceDiscoverer) {
+        return new Runnable() {
+
+            private Set<String> lastInstances;
+
+            @Override
+            public synchronized void run() {
+                try {
+                    Set<String> freshInstances = serviceDiscoverer.getInstances();
+                    if (!(freshInstances.isEmpty() || Objects.equals(lastInstances, freshInstances))) {
+                        lastInstances = freshInstances;
+                        onInstancesRefreshed(lastInstances);
+                    }
+                } catch (Exception ignored) {
+                }
+            }
+        };
+    }
+
     /**
      * Holds operation code and arguments for retry.
      */

src/main/java/org/tarantool/TarantoolClusterClientConfig.java

@@ -6,9 +6,30 @@
  * Configuration for the {@link TarantoolClusterClient}.
  */
 public class TarantoolClusterClientConfig extends TarantoolClientConfig {
-    /* Amount of time (in milliseconds) the operation is eligible for retry. */
-    public int operationExpiryTimeMillis = 500;
 
-    /* Executor service that will be used as a thread of execution to retry writes. */
-    public Executor executor = null;
+    public static final int DEFAULT_OPERATION_EXPIRY_TIME_MILLIS = 500;
+    public static final int DEFAULT_CLUSTER_DISCOVERY_DELAY_MILLIS = 60_000;
+
+    /**
+     * Period for the operation is eligible for retry.
+     */
+    public int operationExpiryTimeMillis = DEFAULT_OPERATION_EXPIRY_TIME_MILLIS;
+
+    /**
+     * Executor that will be used as a thread of
+     * execution to retry writes.
+     */
+    public Executor executor;
+
+    /**
+     * Gets a name of the stored function to be used
+     * to fetch list of instances.
+     */
+    public String clusterDiscoveryEntryFunction;
+
+    /**
+     * Scan period for refreshing a new list of instances.
+     */
+    public int clusterDiscoveryDelayMillis = DEFAULT_CLUSTER_DISCOVERY_DELAY_MILLIS;
+
 }

src/main/java/org/tarantool/TarantoolThreadDaemonFactory.java

@@ -0,0 +1,23 @@
+package org.tarantool;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class TarantoolThreadDaemonFactory implements ThreadFactory {
+
+    private static final AtomicInteger POOL_NUMBER = new AtomicInteger(1);
+    private final AtomicInteger threadNumber = new AtomicInteger(1);
+    private final String namePrefix;
+
+    public TarantoolThreadDaemonFactory(String namePrefix) {
+        this.namePrefix = namePrefix + "-" + POOL_NUMBER.incrementAndGet() + "-thread-";
+    }
+
+    @Override
+    public Thread newThread(Runnable runnable) {
+        Thread thread = new Thread(runnable, namePrefix + threadNumber.incrementAndGet());
+        thread.setDaemon(true);
+
+        return thread;
+    }
+}

src/main/java/org/tarantool/cluster/IllegalDiscoveryFunctionResult.java

@@ -0,0 +1,13 @@
+package org.tarantool.cluster;
+
+/**
+ *  Raised when {@link TarantoolClusterStoredFunctionDiscoverer} validates
+ *  a function result as unsupported.
+ */
+public class IllegalDiscoveryFunctionResult extends RuntimeException {
+
+    public IllegalDiscoveryFunctionResult(String message) {
+        super(message);
+    }
+
+}

src/main/java/org/tarantool/cluster/TarantoolClusterDiscoverer.java

@@ -0,0 +1,21 @@
+package org.tarantool.cluster;
+
+import java.util.Set;
+
+/**
+ * Discovery strategy to obtain a list of the cluster nodes.
+ * This one can be used by {@link org.tarantool.RefreshableSocketProvider}
+ * to provide support for fault tolerance property.
+ *
+ * @see org.tarantool.RefreshableSocketProvider
+ */
+public interface TarantoolClusterDiscoverer {
+
+    /**
+     * Gets nodes addresses in <code>host:port</code> format.
+     *
+     * @return list of the cluster nodes
+     */
+    Set<String> getInstances();
+
+}

src/main/java/org/tarantool/cluster/TarantoolClusterStoredFunctionDiscoverer.java

@@ -0,0 +1,63 @@
+package org.tarantool.cluster;
+
+import org.tarantool.TarantoolClient;
+import org.tarantool.TarantoolClientOps;
+import org.tarantool.TarantoolClusterClientConfig;
+
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * A cluster nodes discoverer based on calling a predefined function
+ * which returns list of nodes.
+ *
+ * The function has to have no arguments and return list of
+ * the strings which follow <code>host[:port]</code> format
+ */
+public class TarantoolClusterStoredFunctionDiscoverer implements TarantoolClusterDiscoverer {
+
+    private TarantoolClient client;
+    private String entryFunction;
+
+    public TarantoolClusterStoredFunctionDiscoverer(TarantoolClusterClientConfig clientConfig, TarantoolClient client) {
+        this.client = client;
+        this.entryFunction = clientConfig.clusterDiscoveryEntryFunction;
+    }
+
+    @Override
+    public Set<String> getInstances() {
+        TarantoolClientOps<Integer, List<?>, Object, List<?>> syncOperations = client.syncOps();
+
+        List<?> list = syncOperations.call(entryFunction);
+        // discoverer expects a single array result from the function now;
+        // in order to protect this contract the discoverer does a strict
+        // validation against the data returned;
+        // this strict-mode allows us to extend the contract in a non-breaking
+        // way for old clients just reserve an extra return value in
+        // terms of LUA multi-result support.
+        checkResult(list);
+
+        List<Object> funcResult = (List<Object>) list.get(0);
+        return funcResult.stream()
+                .map(Object::toString)
+                .collect(Collectors.toCollection(LinkedHashSet::new));
+    }
+
+    /**
+     * Check whether the result follows the contract or not.
+     * The contract is <b>single array of strings</b>
+     *
+     * @param result result to be validated
+     */
+    private void checkResult(List<?> result) {
+        if (result.size() != 1) {
+            throw new IllegalDiscoveryFunctionResult("Too many result values");
+        }
+        if (!((List<Object>)result.get(0)).stream().allMatch(item -> item instanceof String)) {
+            throw new IllegalDiscoveryFunctionResult("The first value must be an array of strings");
+        }
+    }
+
+}

src/main/java/org/tarantool/util/StringUtils.java

@@ -0,0 +1,21 @@
+package org.tarantool.util;
+
+public class StringUtils {
+
+    public static boolean isEmpty(String string) {
+        return (string == null) || (string.isEmpty());
+    }
+
+    public static boolean isNotEmpty(String string) {
+        return !isEmpty(string);
+    }
+
+    public static boolean isBlank(String string) {
+        return (string == null) || (string.trim().isEmpty());
+    }
+
+    public static boolean isNotBlank(String string) {
+        return !isBlank(string);
+    }
+
+}

src/test/java/org/tarantool/AbstractSocketProviderTest.java

@@ -0,0 +1,50 @@
+package org.tarantool;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.nio.channels.SocketChannel;
+import java.util.Collection;
+import java.util.stream.Collectors;
+
+import static org.mockito.Mockito.anyObject;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+public class AbstractSocketProviderTest {
+
+    protected String extractRawHostAndPortString(SocketAddress socketAddress) {
+        InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress;
+        return inetSocketAddress.getAddress().getHostName() + ":" + inetSocketAddress.getPort();
+    }
+
+    protected Iterable<String> asRawHostAndPort(Collection<SocketAddress> addresses) {
+        return addresses.stream()
+                .map(this::extractRawHostAndPortString)
+                .collect(Collectors.toList());
+    }
+
+    protected <T extends BaseSocketChannelProvider> T wrapWithMockChannelProvider(T source) throws IOException {
+        T wrapper = spy(source);
+        doReturn(makeSocketChannel()).when(wrapper).openChannel(anyObject());
+        return wrapper;
+    }
+
+    protected <T extends BaseSocketChannelProvider> T wrapWithMockErroredChannelProvider(T source) throws IOException {
+        T wrapper = spy(source);
+        doThrow(IOException.class).when(wrapper).openChannel(anyObject());
+        return wrapper;
+    }
+
+    private SocketChannel makeSocketChannel() {
+        SocketChannel socketChannel = mock(SocketChannel.class);
+        when(socketChannel.socket()).thenReturn(mock(Socket.class));
+
+        return socketChannel;
+    }
+
+}

src/test/java/org/tarantool/AbstractTarantoolConnectorIT.java

@@ -4,8 +4,8 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.opentest4j.AssertionFailedError;
 
-import java.math.BigInteger;
 import java.io.IOException;
+import java.math.BigInteger;
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.util.List;
@@ -18,13 +18,13 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
-
 import static org.tarantool.TestUtils.makeInstanceEnv;
 
 /**
  * Abstract test. Provides environment control and frequently used functions.
  */
 public abstract class AbstractTarantoolConnectorIT {
+
     protected static final String host = System.getProperty("tntHost", "localhost");
     protected static final int port = Integer.parseInt(System.getProperty("tntPort", "3301"));
     protected static final int consolePort = Integer.parseInt(System.getProperty("tntConsolePort", "3313"));
@@ -37,8 +37,7 @@ public abstract class AbstractTarantoolConnectorIT {
     protected static final int TIMEOUT = 500;
     protected static final int RESTART_TIMEOUT = 2000;
 
-    protected static final SocketChannelProvider socketChannelProvider = new TestSocketChannelProvider(host, port,
-        RESTART_TIMEOUT);
+    protected static final SocketChannelProvider socketChannelProvider = new TestSocketChannelProvider(host, port, RESTART_TIMEOUT);
 
     protected static TarantoolControl control;
     protected static TarantoolConsole console;
@@ -53,36 +52,36 @@ public abstract class AbstractTarantoolConnectorIT {
     protected static int MPK_INDEX_ID;
     protected static int VIDX_INDEX_ID;
 
-    private static final String[] setupScript = new String[] {
-        "box.schema.space.create('basic_test', { format = " +
-            "{{name = 'id', type = 'integer'}," +
-            " {name = 'val', type = 'string'} } })",
+    private static final String[] setupScript = new String[]{
+            "box.schema.space.create('basic_test', { format = " +
+                    "{{name = 'id', type = 'integer'}," +
+                    " {name = 'val', type = 'string'} } })",
 
-        "box.space.basic_test:create_index('pk', { type = 'TREE', parts = {'id'} } )",
-        "box.space.basic_test:create_index('vidx', { type = 'TREE', unique = false, parts = {'val'} } )",
+            "box.space.basic_test:create_index('pk', { type = 'TREE', parts = {'id'} } )",
+            "box.space.basic_test:create_index('vidx', { type = 'TREE', unique = false, parts = {'val'} } )",
 
-        "box.space.basic_test:replace{1, 'one'}",
-        "box.space.basic_test:replace{2, 'two'}",
-        "box.space.basic_test:replace{3, 'three'}",
+            "box.space.basic_test:replace{1, 'one'}",
+            "box.space.basic_test:replace{2, 'two'}",
+            "box.space.basic_test:replace{3, 'three'}",
 
-        "box.schema.space.create('multipart_test', { format = " +
-            "{{name = 'id1', type = 'integer'}," +
-            " {name = 'id2', type = 'string'}," +
-            " {name = 'val1', type = 'string'} } })",
+            "box.schema.space.create('multipart_test', { format = " +
+                    "{{name = 'id1', type = 'integer'}," +
+                    " {name = 'id2', type = 'string'}," +
+                    " {name = 'val1', type = 'string'} } })",
 
-        "box.space.multipart_test:create_index('pk', { type = 'TREE', parts = {'id1', 'id2'} })",
-        "box.space.multipart_test:create_index('vidx', { type = 'TREE', unique = false, parts = {'val1'} })",
+            "box.space.multipart_test:create_index('pk', { type = 'TREE', parts = {'id1', 'id2'} })",
+            "box.space.multipart_test:create_index('vidx', { type = 'TREE', unique = false, parts = {'val1'} })",
 
-        "box.space.multipart_test:replace{1, 'one', 'o n e'}",
-        "box.space.multipart_test:replace{2, 'two', 't w o'}",
-        "box.space.multipart_test:replace{3, 'three', 't h r e e'}",
+            "box.space.multipart_test:replace{1, 'one', 'o n e'}",
+            "box.space.multipart_test:replace{2, 'two', 't w o'}",
+            "box.space.multipart_test:replace{3, 'three', 't h r e e'}",
 
-        "function echo(...) return ... end"
+            "function echo(...) return ... end"
     };
 
-    private static final String[] cleanScript = new String[] {
-        "box.space.basic_test and box.space.basic_test:drop()",
-        "box.space.multipart_test and box.space.multipart_test:drop()"
+    private static final String[] cleanScript = new String[]{
+            "box.space.basic_test and box.space.basic_test:drop()",
+            "box.space.multipart_test and box.space.multipart_test:drop()"
     };
 
     @BeforeAll
@@ -124,7 +123,7 @@ private static void executeLua(String[] exprs) {
     protected void checkTupleResult(Object res, List tuple) {
         assertNotNull(res);
         assertTrue(List.class.isAssignableFrom(res.getClass()));
-        List list = (List)res;
+        List list = (List) res;
         assertEquals(1, list.size());
         assertNotNull(list.get(0));
         assertTrue(List.class.isAssignableFrom(list.get(0).getClass()));
@@ -143,27 +142,27 @@ protected static TarantoolClientConfig makeClientConfig() {
         return fillClientConfig(new TarantoolClientConfig());
     }
 
-    protected static TarantoolClusterClientConfig makeClusterClientConfig() {
+    public static TarantoolClusterClientConfig makeClusterClientConfig() {
         TarantoolClusterClientConfig config = fillClientConfig(new TarantoolClusterClientConfig());
         config.executor = null;
         config.operationExpiryTimeMillis = TIMEOUT;
         return config;
     }
 
-    private static <T> T fillClientConfig(TarantoolClientConfig config) {
+    private static <T extends TarantoolClientConfig> T fillClientConfig(T config) {
         config.username = username;
         config.password = password;
         config.initTimeoutMillis = RESTART_TIMEOUT;
         config.sharedBufferSize = 128;
-        return (T)config;
+        return (T) config;
     }
 
     protected static TarantoolConsole openConsole() {
         return TarantoolConsole.open(host, consolePort);
     }
 
     protected static TarantoolConsole openConsole(String instance) {
-        return TarantoolConsole.open(control.tntCtlWorkDir, instance);
+        return TarantoolConsole.open(TarantoolControl.tntCtlWorkDir, instance);
     }
 
     protected TarantoolConnection openConnection() {
@@ -247,7 +246,7 @@ protected static void startTarantool(String instance) {
      *
      * @param timeout Timeout in ms.
      * @param message Error message.
-     * @param r Runnable.
+     * @param r       Runnable.
      */
     protected void assertTimeoutPreemptively(int timeout, String message, Runnable r) {
         ExecutorService executorService = Executors.newSingleThreadExecutor();

src/test/java/org/tarantool/ClientReconnectClusterIT.java

@@ -0,0 +1,157 @@
+package org.tarantool;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertIterableEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+@DisplayName("A RR socket provider")
+public class RoundRobinSocketProviderImplTest extends AbstractSocketProviderTest {
+
+    @Test
+    @DisplayName("initialized with a right addresses count")
+    public void testAddressesCount() {
+        RoundRobinSocketProviderImpl socketProvider
+                = new RoundRobinSocketProviderImpl("localhost:3301", "127.0.0.1:3302", "10.0.0.10:3303");
+        assertEquals(3, socketProvider.getAddressCount());
+
+        socketProvider.refreshAddresses(Collections.singletonList("10.0.0.1"));
+        assertEquals(1, socketProvider.getAddressCount());
+    }
+
+    @Test
+    @DisplayName("initialized with a right addresses values")
+    public void testAddresses() {
+        String[] addresses = {"localhost:3301", "127.0.0.2:3302", "10.0.0.10:3303"};
+        RoundRobinSocketProviderImpl socketProvider
+                = new RoundRobinSocketProviderImpl(addresses);
+        assertIterableEquals(Arrays.asList(addresses), asRawHostAndPort(socketProvider.getAddresses()));
+
+        List<String> strings = Collections.singletonList("10.0.0.1:3310");
+        socketProvider.refreshAddresses(strings);
+        assertIterableEquals(strings, asRawHostAndPort(socketProvider.getAddresses()));
+    }
+
+    @Test
+    @DisplayName("initialized failed when an empty addresses list is provided")
+    public void testEmptyAddresses() {
+        assertThrows(IllegalArgumentException.class, RoundRobinSocketProviderImpl::new);
+    }
+
+    @Test
+    @DisplayName("changed addresses list with a failure when a new list is empty")
+    public void testResultWithEmptyAddresses() throws IOException {
+        RoundRobinSocketProviderImpl socketProvider
+                = wrapWithMockChannelProvider(new RoundRobinSocketProviderImpl("localhost:3301"));
+
+        assertThrows(IllegalArgumentException.class, () -> socketProvider.refreshAddresses(null));
+        assertThrows(IllegalArgumentException.class, () -> socketProvider.refreshAddresses(Collections.emptyList()));
+    }
+
+    @Test
+    @DisplayName("changed addresses list with a failure when a new list is empty")
+    public void testResultWithWrongAddress() throws IOException {
+        RoundRobinSocketProviderImpl socketProvider
+                = wrapWithMockChannelProvider(new RoundRobinSocketProviderImpl("localhost:3301"));
+
+        assertThrows(IllegalArgumentException.class, () -> socketProvider.refreshAddresses(null));
+        assertThrows(IllegalArgumentException.class, () -> socketProvider.refreshAddresses(Collections.emptyList()));
+    }
+
+    @Test
+    @DisplayName("initialized with a default timeout")
+    public void testDefaultTimeout() {
+        RoundRobinSocketProviderImpl socketProvider
+                = new RoundRobinSocketProviderImpl("localhost");
+        assertEquals(RoundRobinSocketProviderImpl.NO_TIMEOUT, socketProvider.getTimeout());
+    }
+
+    @Test
+    @DisplayName("changed its timeout to new value")
+    public void testChangingTimeout() {
+        RoundRobinSocketProviderImpl socketProvider
+                = new RoundRobinSocketProviderImpl("localhost");
+        int expectedTimeout = 10_000;
+        socketProvider.setTimeout(expectedTimeout);
+        assertEquals(expectedTimeout, socketProvider.getTimeout());
+    }
+
+    @Test
+    @DisplayName("changed to negative timeout with a failure")
+    public void testWrongChangingTimeout() {
+        RoundRobinSocketProviderImpl socketProvider
+                = new RoundRobinSocketProviderImpl("localhost");
+        int negativeValue = -200;
+        assertThrows(IllegalArgumentException.class, () -> socketProvider.setTimeout(negativeValue));
+    }
+
+    @Test
+    @DisplayName("produced socket channels using a ring pool")
+    public void testAddressRingPool() throws IOException {
+        String[] addresses = {"localhost:3301", "10.0.0.1:3302", "10.0.0.2:3309"};
+        RoundRobinSocketProviderImpl socketProvider
+                = wrapWithMockChannelProvider(new RoundRobinSocketProviderImpl(addresses));
+
+        for (int i = 0; i < 27; i++) {
+            socketProvider.get(0, null);
+            assertEquals(addresses[i % 3], extractRawHostAndPortString(socketProvider.getLastObtainedAddress()));
+        }
+    }
+
+    @Test
+    @DisplayName("produced socket channels for the same instance")
+    public void testOneAddressPool() throws IOException {
+        String expectedAddress = "10.0.0.1:3301";
+        String[] addresses = {expectedAddress};
+        RoundRobinSocketProviderImpl socketProvider
+                = wrapWithMockChannelProvider(new RoundRobinSocketProviderImpl(addresses));
+
+        for (int i = 0; i < 5; i++) {
+            socketProvider.get(0, null);
+            assertEquals(expectedAddress, extractRawHostAndPortString(socketProvider.getLastObtainedAddress()));
+        }
+    }
+
+    @Test
+    @DisplayName("produced socket channel with an exception when an attempt number is over")
+    public void testTooManyAttempts() throws IOException {
+        String expectedAddress = "10.0.0.1:3301";
+        String[] addresses = {expectedAddress};
+        RoundRobinSocketProviderImpl socketProvider
+                = wrapWithMockChannelProvider(new RoundRobinSocketProviderImpl(addresses));
+
+        int retriesLimit = 5;
+        socketProvider.setRetriesLimit(retriesLimit);
+
+        for (int i = 0; i < retriesLimit; i++) {
+            socketProvider.get(0, null);
+            assertEquals(expectedAddress, extractRawHostAndPortString(socketProvider.getLastObtainedAddress()));
+        }
+
+        assertThrows(CommunicationException.class, () -> socketProvider.get(retriesLimit, null));
+    }
+
+    @Test
+    @DisplayName("produced a socket channel with a failure when an unreachable address is provided")
+    public void testWrongAddress() throws IOException {
+        RoundRobinSocketProviderImpl socketProvider
+                = wrapWithMockErroredChannelProvider(new RoundRobinSocketProviderImpl("unreachable-host:3301"));
+        assertThrows(CommunicationException.class, () -> socketProvider.get(0, null));
+    }
+
+    @Test
+    @DisplayName("produced a socket channel with a failure with an unreachable address after refresh")
+    public void testWrongRefreshAddress() throws IOException {
+        RoundRobinSocketProviderImpl socketProvider
+                = wrapWithMockErroredChannelProvider(new RoundRobinSocketProviderImpl("unreachable-host:3301"));
+        assertThrows(CommunicationException.class, () -> socketProvider.get(0, null));
+    }
+
+}

src/test/java/org/tarantool/RoundRobinSocketProviderImplTest.java

@@ -0,0 +1,69 @@
+package org.tarantool;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+@DisplayName("A single socket provider")
+class SingleSocketChannelProviderImplTest extends AbstractSocketProviderTest {
+
+    @Test
+    @DisplayName("initialized with a right address")
+    public void testAddressesCount() {
+        String expectedAddress = "localhost:3301";
+        SingleSocketChannelProviderImpl socketProvider
+                = new SingleSocketChannelProviderImpl(expectedAddress);
+        assertEquals(expectedAddress, extractRawHostAndPortString(socketProvider.getAddress()));
+    }
+
+    @Test
+    @DisplayName("poorly initialized with an empty address")
+    public void testEmptyAddresses() {
+        assertThrows(IllegalArgumentException.class, () -> new SingleSocketChannelProviderImpl(null));
+    }
+
+    @Test
+    @DisplayName("initialized with a default timeout")
+    public void testDefaultTimeout() {
+        RoundRobinSocketProviderImpl socketProvider
+                = new RoundRobinSocketProviderImpl("localhost");
+        assertEquals(RoundRobinSocketProviderImpl.NO_TIMEOUT, socketProvider.getTimeout());
+    }
+
+    @Test
+    @DisplayName("changed its timeout to new value")
+    public void testChangingTimeout() {
+        RoundRobinSocketProviderImpl socketProvider
+                = new RoundRobinSocketProviderImpl("localhost");
+        int expectedTimeout = 10_000;
+        socketProvider.setTimeout(expectedTimeout);
+        assertEquals(expectedTimeout, socketProvider.getTimeout());
+    }
+
+    @Test
+    @DisplayName("changed to negative timeout with a failure")
+    public void testWrongChangingTimeout() {
+        RoundRobinSocketProviderImpl socketProvider
+                = new RoundRobinSocketProviderImpl("localhost");
+        int negativeValue = -100;
+        assertThrows(IllegalArgumentException.class, () -> socketProvider.setTimeout(negativeValue));
+    }
+
+    @Test
+    @DisplayName("produced sockets with same address")
+    public void testMultipleChannelGetting() throws IOException {
+        String expectedAddresss = "localhost:3301";
+        SingleSocketChannelProviderImpl socketProvider
+                = wrapWithMockChannelProvider(new SingleSocketChannelProviderImpl(expectedAddresss));
+
+        for (int i = 0; i < 10; i++) {
+            socketProvider.get(0, null);
+            assertEquals(expectedAddresss, extractRawHostAndPortString(socketProvider.getAddress()));
+        }
+    }
+
+}

src/test/java/org/tarantool/SingleSocketChannelProviderImplTest.java

@@ -1,10 +1,28 @@
 package org.tarantool;
 
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 public class TestUtils {
+
+    public static String makeDiscoveryFunction(String functionName, Collection<?> addresses) {
+        String functionResult = addresses.stream()
+                .map(address -> "'" + address + "'")
+                .collect(Collectors.joining(",", "{", "}"));
+        return makeDiscoveryFunction(functionName, functionResult);
+    }
+
+    public static String makeDiscoveryFunction(String functionName, Object result) {
+        return makeDiscoveryFunction(functionName, result.toString());
+    }
+
+    public static String makeDiscoveryFunction(String functionName, String body) {
+        return "function " + functionName + "() return " + body + " end";
+    }
+
     final static String replicationInfoRequest = "return " +
                                                  "box.info.id, " +
                                                  "box.info.lsn, " +

src/test/java/org/tarantool/TestUtils.java

@@ -0,0 +1,179 @@
+package org.tarantool.cluster;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.tarantool.AbstractTarantoolConnectorIT;
+import org.tarantool.CommunicationException;
+import org.tarantool.TarantoolClient;
+import org.tarantool.TarantoolClientImpl;
+import org.tarantool.TarantoolClusterClientConfig;
+import org.tarantool.TarantoolControl;
+import org.tarantool.TarantoolException;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.tarantool.TestUtils.makeDiscoveryFunction;
+import static org.tarantool.TestUtils.makeInstanceEnv;
+
+@DisplayName("A cluster discoverer")
+public class ClusterServiceStoredFunctionDiscovererIT {
+
+    protected static final int INSTANCE_LISTEN_PORT = 3301;
+    protected static final int INSTANCE_ADMIN_PORT = 3313;
+    private static final String LUA_FILE = "jdk-testing.lua";
+
+    private static final String INSTANCE_NAME = "jdk-testing";
+    private static TarantoolControl control;
+
+    private static String ENTRY_FUNCTION_NAME = "getAddresses";
+
+    private TarantoolClusterClientConfig clusterConfig;
+    private TarantoolClient client;
+
+    @BeforeAll
+    public static void setupEnv() {
+        control = new TarantoolControl();
+        control.createInstance(INSTANCE_NAME, LUA_FILE, makeInstanceEnv(INSTANCE_LISTEN_PORT, INSTANCE_ADMIN_PORT));
+
+        control.start(INSTANCE_NAME);
+        control.waitStarted(INSTANCE_NAME);
+    }
+
+    @BeforeEach
+    public void setupTest() {
+        clusterConfig = AbstractTarantoolConnectorIT.makeClusterClientConfig();
+        clusterConfig.clusterDiscoveryEntryFunction = ENTRY_FUNCTION_NAME;
+
+        client = new TarantoolClientImpl("localhost:" + INSTANCE_LISTEN_PORT, clusterConfig);
+    }
+
+    @AfterAll
+    public static void tearDownEnv() {
+        control.stop(INSTANCE_NAME);
+        control.waitStopped(INSTANCE_NAME);
+    }
+
+    @Test
+    @DisplayName("fetched list of addresses")
+    public void testSuccessfulAddressParsing() {
+        List<String> addresses = Arrays.asList("localhost:3311", "127.0.0.1:3301");
+        String functionCode = makeDiscoveryFunction(ENTRY_FUNCTION_NAME, addresses);
+        control.openConsole(INSTANCE_NAME).exec(functionCode);
+
+        TarantoolClusterStoredFunctionDiscoverer discoverer =
+                new TarantoolClusterStoredFunctionDiscoverer(clusterConfig, client);
+
+        Set<String> instances = discoverer.getInstances();
+
+        assertNotNull(instances);
+        assertEquals(2, instances.size());
+        assertTrue(instances.contains(addresses.get(0)));
+        assertTrue(instances.contains(addresses.get(1)));
+    }
+
+    @Test
+    @DisplayName("fetched duplicated addresses")
+    public void testSuccessfulUniqueAddressParsing() {
+        List<String> addresses = Arrays.asList("localhost:3311", "127.0.0.1:3301", "127.0.0.2:3301", "localhost:3311");
+
+        String functionCode = makeDiscoveryFunction(ENTRY_FUNCTION_NAME, addresses);
+        control.openConsole(INSTANCE_NAME).exec(functionCode);
+
+        TarantoolClusterStoredFunctionDiscoverer discoverer =
+                new TarantoolClusterStoredFunctionDiscoverer(clusterConfig, client);
+
+        Set<String> instances = discoverer.getInstances();
+
+        assertNotNull(instances);
+        assertEquals(3, instances.size());
+        assertTrue(instances.contains(addresses.get(0)));
+        assertTrue(instances.contains(addresses.get(1)));
+        assertTrue(instances.contains(addresses.get(3)));
+    }
+
+
+    @Test
+    @DisplayName("fetched empty address list")
+    public void testFunctionReturnedEmptyList() {
+        String functionCode = makeDiscoveryFunction(ENTRY_FUNCTION_NAME, Collections.emptyList());
+        control.openConsole(INSTANCE_NAME).exec(functionCode);
+
+        TarantoolClusterStoredFunctionDiscoverer discoverer =
+                new TarantoolClusterStoredFunctionDiscoverer(clusterConfig, client);
+
+        Set<String> instances = discoverer.getInstances();
+
+        assertNotNull(instances);
+        assertTrue(instances.isEmpty());
+    }
+
+    @Test
+    @DisplayName("fetched with an exception using wrong function name")
+    public void testWrongFunctionName() {
+        clusterConfig.clusterDiscoveryEntryFunction = "wrongFunction";
+
+        TarantoolClusterStoredFunctionDiscoverer discoverer =
+                new TarantoolClusterStoredFunctionDiscoverer(clusterConfig, client);
+
+        assertThrows(TarantoolException.class, discoverer::getInstances);
+    }
+
+    @Test
+    @DisplayName("fetched with an exception using a broken client")
+    public void testWrongInstanceAddress() {
+        clusterConfig.initTimeoutMillis = 1000;
+
+        client.close();
+        TarantoolClusterStoredFunctionDiscoverer discoverer =
+                new TarantoolClusterStoredFunctionDiscoverer(clusterConfig, client);
+
+        assertThrows(CommunicationException.class, discoverer::getInstances);
+    }
+
+    @Test
+    @DisplayName("fetched with an exception when wrong data type returned")
+    public void testWrongTypeResultData() {
+        String functionCode = makeDiscoveryFunction(ENTRY_FUNCTION_NAME, 42);
+        control.openConsole(INSTANCE_NAME).exec(functionCode);
+
+        TarantoolClusterStoredFunctionDiscoverer discoverer =
+                new TarantoolClusterStoredFunctionDiscoverer(clusterConfig, client);
+
+        assertThrows(IllegalDiscoveryFunctionResult.class, discoverer::getInstances);
+    }
+
+    @Test
+    @DisplayName("fetched with an exception when wrong multi result returned")
+    public void testWrongMultiResultData() {
+        String functionCode = makeDiscoveryFunction(ENTRY_FUNCTION_NAME, "{'host1'}, 'host2'");
+        control.openConsole(INSTANCE_NAME).exec(functionCode);
+
+        TarantoolClusterStoredFunctionDiscoverer discoverer =
+                new TarantoolClusterStoredFunctionDiscoverer(clusterConfig, client);
+
+        assertThrows(IllegalDiscoveryFunctionResult.class, discoverer::getInstances);
+    }
+
+    @Test
+    @DisplayName("fetched with an exception using error-prone function")
+    public void testFunctionWithError() {
+        String functionCode = makeDiscoveryFunction(ENTRY_FUNCTION_NAME, "error('msg')");
+        control.openConsole(INSTANCE_NAME).exec(functionCode);
+
+        TarantoolClusterStoredFunctionDiscoverer discoverer =
+                new TarantoolClusterStoredFunctionDiscoverer(clusterConfig, client);
+
+        assertThrows(TarantoolException.class, discoverer::getInstances);
+    }
+
+}