Repository: sysown/proxysql (language: cpp)

Issue:
Fix memory leaks detected during testing
- [x] A clear description of the issue

The following two memory leaks have been detected during testing:

#### First leak

Leak is directly tied to `handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE`,
related to memory allocated to `client_addr`. As we can see by the log:

```
2021-09-10 16:13:55 MySQL_Session.cpp:5098:handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(): [ERROR] ProxySQL Error: Access denied for user 'inv_user'@'10.200.1.2' (using password: YES)
2021-09-10 16:13:56 MySQL_Session.cpp:5098:handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(): [ERROR] ProxySQL Error: Access denied for user 'inv_user'@'10.200.1.2' (using password: YES)
2021-09-10 16:13:56 MySQL_Session.cpp:5098:handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(): [ERROR] ProxySQL Error: Access denied for user 'inv_user'@'10.200.1.2' (using password: YES)
2021-09-10 16:13:57 MySQL_Session.cpp:5098:handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(): [ERROR] ProxySQL Error: Access denied for user 'inv_user'@'10.200.1.2' (using password: YES)
```

```
==1202933==ERROR: LeakSanitizer: detected memory leaks

Direct leak of 44 byte(s) in 4 object(s) allocated from:
    #0 0x7f1340b783f9 in __interceptor_strdup /build/gcc/src/gcc/libsanitizer/asan/asan_interceptors.cpp:454
    #1 0x565348b76756 in MySQL_Session::handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(_PtrSize_t*, bool*) /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Session.cpp:5059
    #2 0x565348b61e4a in MySQL_Session::get_pkts_from_client(bool&, _PtrSize_t&) /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Session.cpp:3416
    #3 0x565348b6bdd1 in MySQL_Session::handler() /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Session.cpp:4254
    #4 0x565348aeb2a0 in MySQL_Thread::process_all_sessions() /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Thread.cpp:3539
    #5 0x565348ae51a3 in MySQL_Thread::run() /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Thread.cpp:3051
    #6 0x565348902f6e in mysql_worker_thread_func(void*) /home/javjarfer/Projects/proxysql_v2.X.0/src/main.cpp:417
    #7 0x7f1340903258 in start_thread (/usr/lib/libpthread.so.0+0x9258)

Direct leak of 36 byte(s) in 1 object(s) allocated from:
    #0 0x7f1340bd3279 in __interceptor_malloc /build/gcc/src/gcc/libsanitizer/asan/asan_malloc_linux.cpp:145
    #1 0x565348aaf5d4 in MySQL_Data_Stream::buffer2array() /home/javjarfer/Projects/proxysql_v2.X.0/lib/mysql_data_stream.cpp:975
    #2 0x565348aad8ff in MySQL_Data_Stream::read_pkts() /home/javjarfer/Projects/proxysql_v2.X.0/lib/mysql_data_stream.cpp:892
    #3 0x565348c36fd5 in child_mysql(void*) /home/javjarfer/Projects/proxysql_v2.X.0/lib/ProxySQL_Admin.cpp:4908
    #4 0x7f1340903258 in start_thread (/usr/lib/libpthread.so.0+0x9258)
```

Leak is related with failed connections. Same result isn't achieved for non-failing
connections.

#### Second leak

Leak is related with SSL connections:

```
==1199044==ERROR: LeakSanitizer: detected memory leaks

Direct leak of 252 byte(s) in 7 object(s) allocated from:
    #0 0x7f3fb412b279 in __interceptor_malloc /build/gcc/src/gcc/libsanitizer/asan/asan_malloc_linux.cpp:145
    #1 0x5612e29f05d4 in MySQL_Data_Stream::buffer2array() /home/javjarfer/Projects/proxysql_v2.X.0/lib/mysql_data_stream.cpp:975
    #2 0x5612e29ee8ff in MySQL_Data_Stream::read_pkts() /home/javjarfer/Projects/proxysql_v2.X.0/lib/mysql_data_stream.cpp:892
    #3 0x5612e2a28fa7 in MySQL_Thread::process_data_on_data_stream(MySQL_Data_Stream*, unsigned int) /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Thread.cpp:3277
    #4 0x5612e2a23dd0 in MySQL_Thread::ProcessAllMyDS_AfterPoll() /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Thread.cpp:2793
    #5 0x5612e2a25f8d in MySQL_Thread::run() /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Thread.cpp:3035
    #6 0x5612e2843f6e in mysql_worker_thread_func(void*) /home/javjarfer/Projects/proxysql_v2.X.0/src/main.cpp:417
    #7 0x7f3fb3e5b258 in start_thread (/usr/lib/libpthread.so.0+0x9258)

Direct leak of 36 byte(s) in 1 object(s) allocated from:
    #0 0x7f3fb412b279 in __interceptor_malloc /build/gcc/src/gcc/libsanitizer/asan/asan_malloc_linux.cpp:145
    #1 0x5612e29f05d4 in MySQL_Data_Stream::buffer2array() /home/javjarfer/Projects/proxysql_v2.X.0/lib/mysql_data_stream.cpp:975
    #2 0x5612e29ee8ff in MySQL_Data_Stream::read_pkts() /home/javjarfer/Projects/proxysql_v2.X.0/lib/mysql_data_stream.cpp:892
    #3 0x5612e2b77fd5 in child_mysql(void*) /home/javjarfer/Projects/proxysql_v2.X.0/lib/ProxySQL_Admin.cpp:4908
    #4 0x7f3fb3e5b258 in start_thread (/usr/lib/libpthread.so.0+0x9258)
```

- [x] ProxySQL version

v2.x

- [x] OS version

```
NAME="Arch Linux"
PRETTY_NAME="Arch Linux"
ID=arch
BUILD_ID=rolling
```

- [x] The steps to reproduce the issue

The provided script should be enough to reproduce both issues, sanitizer output should be:

```
Direct leak of 108 byte(s) in 3 object(s) allocated from:
    #0 0x7f1985b0d279 in __interceptor_malloc /build/gcc/src/gcc/libsanitizer/asan/asan_malloc_linux.cpp:145
    #1 0x56403038a5d4 in MySQL_Data_Stream::buffer2array() /home/javjarfer/Projects/proxysql_v2.X.0/lib/mysql_data_stream.cpp:975
    #2 0x5640303888ff in MySQL_Data_Stream::read_pkts() /home/javjarfer/Projects/proxysql_v2.X.0/lib/mysql_data_stream.cpp:892
    #3 0x5640303c2fa7 in MySQL_Thread::process_data_on_data_stream(MySQL_Data_Stream*, unsigned int) /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Thread.cpp:3277
    #4 0x5640303bddd0 in MySQL_Thread::ProcessAllMyDS_AfterPoll() /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Thread.cpp:2793
    #5 0x5640303bff8d in MySQL_Thread::run() /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Thread.cpp:3035
    #6 0x5640301ddf6e in mysql_worker_thread_func(void*) /home/javjarfer/Projects/proxysql_v2.X.0/src/main.cpp:417
    #7 0x7f198583d258 in start_thread (/usr/lib/libpthread.so.0+0x9258)

Direct leak of 108 byte(s) in 3 object(s) allocated from:
    #0 0x7f1985b0d279 in __interceptor_malloc /build/gcc/src/gcc/libsanitizer/asan/asan_malloc_linux.cpp:145
    #1 0x56403038a5d4 in MySQL_Data_Stream::buffer2array() /home/javjarfer/Projects/proxysql_v2.X.0/lib/mysql_data_stream.cpp:975
    #2 0x5640303888ff in MySQL_Data_Stream::read_pkts() /home/javjarfer/Projects/proxysql_v2.X.0/lib/mysql_data_stream.cpp:892
    #3 0x564030511fd5 in child_mysql(void*) /home/javjarfer/Projects/proxysql_v2.X.0/lib/ProxySQL_Admin.cpp:4908
    #4 0x7f198583d258 in start_thread (/usr/lib/libpthread.so.0+0x9258)

Direct leak of 33 byte(s) in 3 object(s) allocated from:
    #0 0x7f1985ab23f9 in __interceptor_strdup /build/gcc/src/gcc/libsanitizer/asan/asan_interceptors.cpp:454
    #1 0x564030451756 in MySQL_Session::handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(_PtrSize_t*, bool*) /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Session.cpp:5059
    #2 0x56403043ce4a in MySQL_Session::get_pkts_from_client(bool&, _PtrSize_t&) /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Session.cpp:3416
    #3 0x564030446dd1 in MySQL_Session::handler() /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Session.cpp:4254
    #4 0x5640303c62a0 in MySQL_Thread::process_all_sessions() /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Thread.cpp:3539
    #5 0x5640303c01a3 in MySQL_Thread::run() /home/javjarfer/Projects/proxysql_v2.X.0/lib/MySQL_Thread.cpp:3051
    #6 0x5640301ddf6e in mysql_worker_thread_func(void*) /home/javjarfer/Projects/proxysql_v2.X.0/src/main.cpp:417
    #7 0x7f198583d258 in start_thread (/usr/lib/libpthread.so.0+0x9258)
```

Script: 
[reproduce_leaks.sh.tar.gz](https://github.com/sysown/proxysql/files/7144577/reproduce_leaks.sh.tar.gz)




Discussion / hints:

Fix patch:
diff --git a/include/MySQL_HostGroups_Manager.h b/include/MySQL_HostGroups_Manager.h
index 6f74e142e8..cc88dfcc87 100644
--- a/include/MySQL_HostGroups_Manager.h
+++ b/include/MySQL_HostGroups_Manager.h
@@ -417,6 +417,8 @@ class MySQL_HostGroups_Manager {
 	void p_update_connection_pool_update_counter(std::string& endpoint_id, std::map<std::string, std::string> labels, std::map<std::string, prometheus::Counter*>& m_map, unsigned long long value, p_hg_dyn_counter::metric idx);
 	void p_update_connection_pool_update_gauge(std::string& endpoint_id, std::map<std::string, std::string> labels, std::map<std::string, prometheus::Gauge*>& m_map, unsigned long long value, p_hg_dyn_gauge::metric idx);
 
+	void group_replication_lag_action_set_server_status(MyHGC* myhgc, char* address, int port, int lag_count, bool enable);
+
 	public:
 	std::mutex galera_set_writer_mutex;
 	pthread_rwlock_t gtid_rwlock;
@@ -558,7 +560,30 @@ class MySQL_HostGroups_Manager {
 	void update_group_replication_set_read_only(char *_hostname, int _port, int _writer_hostgroup, char *error);
 	void update_group_replication_set_writer(char *_hostname, int _port, int _writer_hostgroup);
 	void converge_group_replication_config(int _writer_hostgroup);
-
+	/**
+	 * @brief Set the supplied server as SHUNNED, this function shall be called
+	 *   to 'SHUNNED' those servers which replication lag is bigger than:
+	 *     - `mysql_thread___monitor_groupreplication_max_transactions_behind_count`
+	 *
+	 * @details The function automatically handles the appropriate operation to
+	 *   perform on the supplied server, based on the supplied 'enable' flag and
+	 *   in 'monitor_groupreplication_max_transaction_behind_for_read_only'
+	 *   variable. In case the value of the variable is:
+	 *
+	 *     * '0' or '2': It's required to search the writer hostgroup for
+	 *       finding the supplied server.
+	 *     * '1' or '2': It's required to search the reader hostgroup for
+	 *       finding the supplied server.
+	 *
+	 * @param _hid The writer hostgroup.
+	 * @param address The server address.
+	 * @param port The server port.
+	 * @param lag_counts The computed lag for the sever.
+	 * @param read_only Boolean specifying the read_only flag value of the server.
+	 * @param enable Boolean specifying if the server needs to be disabled / enabled,
+	 *   'true' for enabling the server if it's 'SHUNNED', 'false' for disabling it.
+	 */
+	void group_replication_lag_action(int _hid, char *address, unsigned int port, int lag_counts, bool read_only, bool enable);
 	void update_galera_set_offline(char *_hostname, int _port, int _writer_hostgroup, char *error, bool soft=false);
 	void update_galera_set_read_only(char *_hostname, int _port, int _writer_hostgroup, char *error);
 	void update_galera_set_writer(char *_hostname, int _port, int _writer_hostgroup);
diff --git a/include/MySQL_Thread.h b/include/MySQL_Thread.h
index 35d3e2f188..308f6b8f85 100644
--- a/include/MySQL_Thread.h
+++ b/include/MySQL_Thread.h
@@ -118,6 +118,7 @@ enum MySQL_Thread_status_variable {
 	st_var_aws_aurora_replicas_skipped_during_query,
 	st_var_automatic_detected_sqli,
 	st_var_whitelisted_sqli_fingerprint,
+	st_var_client_host_error_killed_connections,
 	st_var_END
 };
 
@@ -320,6 +321,7 @@ struct p_th_counter {
 		whitelisted_sqli_fingerprint,
 		mysql_killed_backend_connections,
 		mysql_killed_backend_queries,
+		client_host_error_killed_connections,
 		__size
 	};
 };
@@ -359,6 +361,20 @@ struct th_metrics_map_idx {
 	};
 };
 
+/**
+ * @brief Structure holding the data for a Client_Host_Cache entry.
+ */
+typedef struct _MySQL_Client_Host_Cache_Entry {
+	/**
+	 * @brief Last time the entry was updated.
+	 */
+	uint64_t updated_at;
+	/**
+	 * @brief Error count associated with the entry.
+	 */
+	uint32_t error_count;
+} MySQL_Client_Host_Cache_Entry;
+
 class MySQL_Threads_Handler
 {
 	private:
@@ -382,6 +398,22 @@ class MySQL_Threads_Handler
 	//   variable address
 	//   special variable : if true, further input validation is required
 	std::unordered_map<std::string, std::tuple<bool *, bool>> VariablesPointers_bool;
+	/**
+	 * @brief Holds the clients host cache. It keeps track of the number of
+	 *   errors associated to a specific client:
+	 *     - Key: client identifier, based on 'clientaddr'.
+	 *     - Value: Structure of type 'MySQL_Client_Host_Cache_Entry' holding
+	 *       the last time the entry was updated and the error count associated
+	 *       with the client.
+	 */
+	std::unordered_map<std::string, MySQL_Client_Host_Cache_Entry> client_host_cache;
+	/**
+	 * @brief Holds the mutex for accessing 'client_host_cache', since every
+	 *   access can potentially perform 'read/write' operations, a regular mutex
+	 *   is enough.
+	 */
+	pthread_mutex_t mutex_client_host_cache;
+
 	public:
 	struct {
 		int monitor_history;
@@ -410,6 +442,7 @@ class MySQL_Threads_Handler
 		int monitor_groupreplication_healthcheck_timeout;
 		int monitor_groupreplication_healthcheck_max_timeout_count;
 		int monitor_groupreplication_max_transactions_behind_count;
+		int monitor_groupreplication_max_transactions_behind_for_read_only;
 		int monitor_galera_healthcheck_interval;
 		int monitor_galera_healthcheck_timeout;
 		int monitor_galera_healthcheck_max_timeout_count;
@@ -429,6 +462,8 @@ class MySQL_Threads_Handler
 		int query_retries_on_failure;
 		bool client_multi_statements;
 		bool connection_warming;
+		int client_host_cache_size;
+		int client_host_error_counts;
 		int connect_retries_on_failure;
 		int connect_retries_delay;
 		int connection_delay_multiplex_ms;
@@ -548,6 +583,77 @@ class MySQL_Threads_Handler
 		std::array<prometheus::Counter*, p_th_counter::__size> p_counter_array {};
 		std::array<prometheus::Gauge*, p_th_gauge::__size> p_gauge_array {};
 	} status_variables;
+	/**
+	 * @brief Update the client host cache with the supplied 'client_sockaddr',
+	 *   and the supplied 'error' parameter specifying if there was a connection
+	 *   error or not.
+	 *
+	 *   NOTE: This function is not safe, the supplied 'client_sockaddr' should
+	 *   have been initialized by 'accept' or 'getpeername'. NULL checks are not
+	 *   performed.
+	 *
+	 * @details The 'client_sockaddr' parameter is inspected, and the
+	 *   'client_host_cache' map is only updated in case of:
+	 *    - 'address_family' is either 'AF_INET' or 'AF_INET6'.
+	 *    - The address obtained from it isn't '127.0.0.1'.
+	 *
+	 *   In case 'client_sockaddr' matches the previous description, the update
+	 *   of the client host cache is performed in the following way:
+	 *     1. If the cache is full, the oldest element in the cache is searched.
+	 *     In case the oldest element address doesn't match the supplied
+	 *     address, the oldest element is removed.
+	 *     2. The cache is searched looking for the supplied address, in case of
+	 *     being found, the entry is updated, otherwise the entry is inserted in
+	 *     the cache.
+	 *
+	 * @param client_sockaddr A 'sockaddr' holding the required client information
+	 *   to update the 'client_host_cache_map'.
+	 * @param error 'true' if there was an error in the connection that should be
+	 *   register, 'false' otherwise.
+	 */
+	void update_client_host_cache(struct sockaddr* client_sockaddr, bool error);
+	/**
+	 * @brief Retrieves the entry of the underlying 'client_host_cache' map for
+	 *   the supplied 'client_sockaddr' in case of existing. In case it doesn't
+	 *   exist or the supplied 'client_sockaddr' doesn't met the requirements
+	 *   for being registered in the map, and zeroed 'MySQL_Client_Host_Cache_Entry'
+	 *   is returned.
+	 *
+	 *   NOTE: This function is not safe, the supplied 'client_sockaddr' should
+	 *   have been initialized by 'accept' or 'getpeername'. NULL checks are not
+	 *   performed.
+	 *
+	 * @details The 'client_sockaddr' parameter is inspected, and the
+	 *   'client_host_cache' map is only searched in case of:
+	 *    - 'address_family' is either 'AF_INET' or 'AF_INET6'.
+	 *    - The address obtained from it isn't '127.0.0.1'.
+	 *
+	 * @param client_sockaddr A 'sockaddr' holding the required client information
+	 *   to update the 'client_host_cache_map'.
+	 * @return If found, the corresponding entry for the supplied 'client_sockaddr',
+	 *   a zeroed 'MySQL_Client_Host_Cache_Entry' otherwise.
+	 */
+	MySQL_Client_Host_Cache_Entry find_client_host_cache(struct sockaddr* client_sockaddr);
+	/**
+	 * @brief Delete all the entries in the 'client_host_cache' internal map.
+	 */
+	void flush_client_host_cache();
+	/**
+	 * @brief Returns the current entries of 'client_host_cache' in a
+	 *   'SQLite3_result'. In case the param 'reset' is specified, the structure
+	 *   is cleaned after being queried.
+	 *
+	 * @param reset If 'true' the entries of the internal structure
+	 *   'client_host_cache' will be cleaned after scrapping.
+	 *
+	 * @return SQLite3_result holding the current entries of the
+	 *   'client_host_cache'. In the following format:
+	 *
+	 *    [ 'client_address', 'error_num', 'last_updated' ]
+	 *
+	 *    Where 'last_updated' is the last updated time expressed in 'ns'.
+	 */
+	SQLite3_result* get_client_host_cache(bool reset);
 	/**
 	 * @brief Callback to update the metrics.
 	 */
diff --git a/include/SQLite3_Server.h b/include/SQLite3_Server.h
index 288d011567..1c26d5f337 100644
--- a/include/SQLite3_Server.h
+++ b/include/SQLite3_Server.h
@@ -13,6 +13,10 @@ class SQLite3_Session {
 	~SQLite3_Session();
 };
 
+#ifdef TEST_GROUPREP
+using group_rep_status = std::tuple<bool, bool, uint32_t>;
+#endif
+
 class SQLite3_Server {
 	private:
 	volatile int main_shutdown;
@@ -43,6 +47,7 @@ class SQLite3_Server {
 	std::vector<table_def_t *> *tables_defs_galera;
 #endif // TEST_GALERA
 #ifdef TEST_GROUPREP
+	std::unordered_map<std::string, group_rep_status> grouprep_map;
 	std::vector<table_def_t *> *tables_defs_grouprep;
 #endif // TEST_GROUPREP
 #if defined(TEST_AURORA) || defined(TEST_GALERA) || defined(TEST_GROUPREP)
@@ -69,9 +74,11 @@ class SQLite3_Server {
 	void init_galera_ifaces_string(std::string& s);
 #endif // TEST_GALERA
 #ifdef TEST_GROUPREP
+	unsigned int max_num_grouprep_servers;
 	pthread_mutex_t grouprep_mutex;
 	void populate_grouprep_table(MySQL_Session *sess, int txs_behind = 0);
 	void init_grouprep_ifaces_string(std::string& s);
+	group_rep_status grouprep_test_value(const std::string& srv_addr);
 #endif // TEST_GROUPREP
 	SQLite3_Server();
 	~SQLite3_Server();
diff --git a/include/proxysql_admin.h b/include/proxysql_admin.h
index 421715b9e2..54927da24f 100644
--- a/include/proxysql_admin.h
+++ b/include/proxysql_admin.h
@@ -355,6 +355,7 @@ class ProxySQL_Admin {
 	void stats___proxysql_servers_metrics();
 	void stats___mysql_prepared_statements_info();
 	void stats___mysql_gtid_executed();
+	void stats___mysql_client_host_cache(bool reset);
 
 	// Update prometheus metrics
 	void p_stats___memory_metrics();
diff --git a/include/proxysql_structs.h b/include/proxysql_structs.h
index 32dc9510f0..a6279a879e 100644
--- a/include/proxysql_structs.h
+++ b/include/proxysql_structs.h
@@ -799,6 +799,8 @@ __thread bool mysql_thread___enable_client_deprecate_eof;
 __thread bool mysql_thread___enable_server_deprecate_eof;
 __thread bool mysql_thread___log_mysql_warnings_enabled;
 __thread bool mysql_thread___enable_load_data_local_infile;
+__thread int mysql_thread___client_host_cache_size;
+__thread int mysql_thread___client_host_error_counts;
 
 /* variables used for Query Cache */
 __thread int mysql_thread___query_cache_size_MB;
@@ -842,6 +844,7 @@ __thread int mysql_thread___monitor_groupreplication_healthcheck_interval;
 __thread int mysql_thread___monitor_groupreplication_healthcheck_timeout;
 __thread int mysql_thread___monitor_groupreplication_healthcheck_max_timeout_count;
 __thread int mysql_thread___monitor_groupreplication_max_transactions_behind_count;
+__thread int mysql_thread___monitor_groupreplication_max_transaction_behind_for_read_only;
 __thread int mysql_thread___monitor_galera_healthcheck_interval;
 __thread int mysql_thread___monitor_galera_healthcheck_timeout;
 __thread int mysql_thread___monitor_galera_healthcheck_max_timeout_count;
@@ -954,6 +957,8 @@ extern __thread bool mysql_thread___enable_client_deprecate_eof;
 extern __thread bool mysql_thread___enable_server_deprecate_eof;
 extern __thread bool mysql_thread___log_mysql_warnings_enabled;
 extern __thread bool mysql_thread___enable_load_data_local_infile;
+extern __thread int mysql_thread___client_host_cache_size;
+extern __thread int mysql_thread___client_host_error_counts;
 
 /* variables used for Query Cache */
 extern __thread int mysql_thread___query_cache_size_MB;
@@ -996,6 +1001,7 @@ extern __thread int mysql_thread___monitor_replication_lag_count;
 extern __thread int mysql_thread___monitor_groupreplication_healthcheck_interval;
 extern __thread int mysql_thread___monitor_groupreplication_healthcheck_timeout;
 extern __thread int mysql_thread___monitor_groupreplication_healthcheck_max_timeout_count;
+extern __thread int mysql_thread___monitor_groupreplication_max_transaction_behind_for_read_only;
 extern __thread int mysql_thread___monitor_groupreplication_max_transactions_behind_count;
 extern __thread int mysql_thread___monitor_galera_healthcheck_interval;
 extern __thread int mysql_thread___monitor_galera_healthcheck_timeout;
diff --git a/lib/MySQL_HostGroups_Manager.cpp b/lib/MySQL_HostGroups_Manager.cpp
index 5fe9c20773..21c64391cc 100644
--- a/lib/MySQL_HostGroups_Manager.cpp
+++ b/lib/MySQL_HostGroups_Manager.cpp
@@ -16,6 +16,7 @@
 #include <prometheus/gauge.h>
 
 #include "prometheus_helpers.h"
+#include "proxysql_utils.h"
 
 #define char_malloc (char *)malloc
 #define itostr(__s, __i)  { __s=char_malloc(32); sprintf(__s, "%lld", __i); }
@@ -3374,6 +3375,114 @@ void MySQL_HostGroups_Manager::replication_lag_action(int _hid, char *address, u
 	GloAdmin->mysql_servers_wrunlock();
 }
 
+/**
+ * @brief Finds the supplied server in the provided 'MyHGC' and sets the status
+ *   either to 'MYSQL_SERVER_STATUS_SHUNNED_REPLICATION_LAG' if 'enable' is
+ *   'false' or 'MYSQL_SERVER_STATUS_ONLINE' if 'true'. If either of the
+ *   'myhgc' or 'address' params are 'NULL' the function performs no action,
+ *   and returns immediately.
+ *
+ * @param myhgc The MySQL Hostgroup Container in which to perform the server
+ *   search.
+ * @param address The server address.
+ * @param port The server port.
+ * @param lag_count The lag count, computed by 'get_lag_behind_count'.
+ * @param enable Boolean specifying if the server should be enabled or not.
+ */
+void MySQL_HostGroups_Manager::group_replication_lag_action_set_server_status(MyHGC* myhgc, char* address, int port, int lag_count, bool enable) {
+	if (myhgc == NULL || address == NULL) return;
+
+	for (int j=0; j<(int)myhgc->mysrvs->cnt(); j++) {
+		MySrvC *mysrvc=(MySrvC *)myhgc->mysrvs->servers->index(j);
+		if (strcmp(mysrvc->address,address)==0 && mysrvc->port==port) {
+
+			if (enable == true) {
+				if (mysrvc->status==MYSQL_SERVER_STATUS_SHUNNED_REPLICATION_LAG || mysrvc->status==MYSQL_SERVER_STATUS_SHUNNED) {
+					mysrvc->status=MYSQL_SERVER_STATUS_ONLINE;
+					proxy_info("Re-enabling server %u:%s:%d from replication lag\n", myhgc->hid, address, port);
+				}
+			} else {
+				if (mysrvc->status==MYSQL_SERVER_STATUS_ONLINE) {
+					proxy_warning("Shunning 'soft' server %u:%s:%d with replication lag, count number: %d\n", myhgc->hid, address, port, lag_count);
+					mysrvc->status=MYSQL_SERVER_STATUS_SHUNNED;
+				} else {
+					if (mysrvc->status==MYSQL_SERVER_STATUS_SHUNNED) {
+						if (lag_count >= ( mysql_thread___monitor_groupreplication_max_transactions_behind_count * 2 )) {
+							proxy_warning("Shunning 'hard' server %u:%s:%d with replication lag, count number: %d\n", myhgc->hid, address, port, lag_count);
+							mysrvc->status=MYSQL_SERVER_STATUS_SHUNNED_REPLICATION_LAG;
+						}
+					}
+				}
+			}
+		}
+	}
+}
+
+void MySQL_HostGroups_Manager::group_replication_lag_action(
+	int _hid, char *address, unsigned int port, int lag_counts, bool read_only, bool enable
+) {
+	GloAdmin->mysql_servers_wrlock();
+	wrlock();
+
+	int reader_hostgroup = 0;
+	bool writer_is_also_reader = false;
+
+	// Get the reader_hostgroup for the supplied writter hostgroup
+	std::string t_reader_hostgroup_query {
+		"SELECT reader_hostgroup,writer_is_also_reader FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d"
+	};
+	std::string reader_hostgroup_query {};
+	string_format(t_reader_hostgroup_query, reader_hostgroup_query, _hid);
+
+	int cols=0;
+	char *error=NULL;
+	int affected_rows=0;
+	SQLite3_result* rhid_res=NULL;
+	SQLite3_row* rhid_row=nullptr;
+
+	mydb->execute_statement(
+		reader_hostgroup_query.c_str(), &error , &cols , &affected_rows , &rhid_res
+	);
+
+	// If the server isn't present in the supplied hostgroup, there is nothing to do.
+	if (rhid_res->rows.empty() || rhid_res->rows[0]->get_size() == 0) {
+		goto __exit_replication_lag_action;
+	}
+
+	rhid_row = rhid_res->rows[0];
+	reader_hostgroup = atoi(rhid_row->fields[0]);
+	writer_is_also_reader = atoi(rhid_row->fields[1]);
+
+	{
+		MyHGC* myhgc = nullptr;
+
+		if (
+			mysql_thread___monitor_groupreplication_max_transaction_behind_for_read_only == 0 ||
+			mysql_thread___monitor_groupreplication_max_transaction_behind_for_read_only == 2 ||
+			enable
+		) {
+			if (read_only == false) {
+				myhgc = MyHGM->MyHGC_find(_hid);
+				group_replication_lag_action_set_server_status(myhgc, address, port, lag_counts, enable);
+			}
+		}
+
+		if (
+			mysql_thread___monitor_groupreplication_max_transaction_behind_for_read_only == 1 ||
+			mysql_thread___monitor_groupreplication_max_transaction_behind_for_read_only == 2 ||
+			enable
+		) {
+			myhgc = MyHGM->MyHGC_find(reader_hostgroup);
+			group_replication_lag_action_set_server_status(myhgc, address, port, lag_counts, enable);
+		}
+	}
+
+__exit_replication_lag_action:
+
+	wrunlock();
+	GloAdmin->mysql_servers_wrunlock();
+}
+
 void MySQL_HostGroups_Manager::drop_all_idle_connections() {
 	// NOTE: the caller should hold wrlock
 	int i, j;
@@ -4529,19 +4638,31 @@ void MySQL_HostGroups_Manager::update_group_replication_set_offline(char *_hostn
 			GloAdmin->mysql_servers_wrlock();
 			mydb->execute("DELETE FROM mysql_servers_incoming");
 			mydb->execute("INSERT INTO mysql_servers_incoming SELECT hostgroup_id, hostname, port, gtid_port, weight, status, compression, max_connections, max_replication_lag, use_ssl, max_latency_ms, comment FROM mysql_servers");
-			q=(char *)"UPDATE OR IGNORE mysql_servers_incoming SET hostgroup_id=(SELECT offline_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d) WHERE hostname='%s' AND port=%d AND hostgroup_id<>(SELECT offline_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d)";
+			// NOTE: Only updated the servers that have belong to the same cluster.
+			q=(char *)"UPDATE OR IGNORE mysql_servers_incoming SET hostgroup_id=(SELECT offline_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d) WHERE hostname='%s' AND port=%d AND hostgroup_id IN ("
+				" SELECT %d UNION ALL"
+				" SELECT backup_writer_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d UNION ALL"
+				" SELECT reader_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d"
+			")";
 			query=(char *)malloc(strlen(q)+strlen(_hostname)+64);
-			sprintf(query,q,_writer_hostgroup,_hostname,_port,_writer_hostgroup);
+			sprintf(query,q,_writer_hostgroup,_hostname,_port,_writer_hostgroup,_writer_hostgroup,_writer_hostgroup);
 			mydb->execute(query);
-			//free(query);
-			q=(char *)"DELETE FROM mysql_servers_incoming WHERE hostname='%s' AND port=%d AND hostgroup_id<>(SELECT offline_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d)";
-			//query=(char *)malloc(strlen(q)+strlen(_hostname)+64);
-			sprintf(query,q,_hostname,_port,_writer_hostgroup);
+			// NOTE: Only delete the servers that have belong to the same cluster.
+			q=(char*)"DELETE FROM mysql_servers_incoming WHERE hostname='%s' AND port=%d AND hostgroup_id IN ("
+				" SELECT %d UNION ALL"
+				" SELECT backup_writer_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d UNION ALL"
+				" SELECT reader_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d"
+			")";
+			sprintf(query,q,_hostname,_port,_writer_hostgroup,_writer_hostgroup,_writer_hostgroup);
 			mydb->execute(query);
 			//free(query);
-			q=(char *)"UPDATE mysql_servers_incoming SET status=0 WHERE hostname='%s' AND port=%d AND hostgroup_id=(SELECT offline_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d)";
-			//query=(char *)malloc(strlen(q)+strlen(_hostname)+64);
-			sprintf(query,q,_hostname,_port,_writer_hostgroup);
+			// q=(char *)"UPDATE mysql_servers_incoming SET status=0 WHERE hostname='%s' AND port=%d AND hostgroup_id=(SELECT offline_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d)";
+			// sprintf(query,q,_hostname,_port,_writer_hostgroup);
+			q=(char *)"UPDATE mysql_servers_incoming SET status=(CASE "
+				" (SELECT status FROM mysql_servers_incoming WHERE hostname='%s' AND port=%d AND"
+					" hostgroup_id=(SELECT offline_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d)) WHEN 2 THEN 2 ELSE 0 END)"
+				" WHERE hostname='%s' AND port=%d AND hostgroup_id=(SELECT offline_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d)";
+			sprintf(query,q,_hostname,_port,_writer_hostgroup,_hostname,_port,_writer_hostgroup);
 			mydb->execute(query);
 			//free(query);
 			converge_group_replication_config(_writer_hostgroup);
@@ -4605,19 +4726,30 @@ void MySQL_HostGroups_Manager::update_group_replication_set_read_only(char *_hos
 			GloAdmin->mysql_servers_wrlock();
 			mydb->execute("DELETE FROM mysql_servers_incoming");
 			mydb->execute("INSERT INTO mysql_servers_incoming SELECT hostgroup_id, hostname, port, gtid_port, weight, status, compression, max_connections, max_replication_lag, use_ssl, max_latency_ms, comment FROM mysql_servers");
-			q=(char *)"UPDATE OR IGNORE mysql_servers_incoming SET hostgroup_id=(SELECT reader_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d) WHERE hostname='%s' AND port=%d AND hostgroup_id<>(SELECT reader_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d)";
+			// NOTE: Only updated the servers that have belong to the same cluster.
+			q=(char *)"UPDATE OR IGNORE mysql_servers_incoming SET hostgroup_id=(SELECT reader_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d) WHERE hostname='%s' AND port=%d AND hostgroup_id IN ("
+				" SELECT %d UNION ALL"
+				" SELECT backup_writer_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d UNION ALL"
+				" SELECT offline_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d"
+			")";
 			query=(char *)malloc(strlen(q)+strlen(_hostname)+64);
-			sprintf(query,q,_writer_hostgroup,_hostname,_port,_writer_hostgroup);
+			sprintf(query,q,_writer_hostgroup,_hostname,_port,_writer_hostgroup,_writer_hostgroup,_writer_hostgroup);
 			mydb->execute(query);
-			//free(query);
-			q=(char *)"DELETE FROM mysql_servers_incoming WHERE hostname='%s' AND port=%d AND hostgroup_id<>(SELECT reader_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d)";
-			//query=(char *)malloc(strlen(q)+strlen(_hostname)+64);
-			sprintf(query,q,_hostname,_port,_writer_hostgroup);
+			// NOTE: Only delete the servers that have belong to the same cluster.
+			q=(char*)"DELETE FROM mysql_servers_incoming WHERE hostname='%s' AND port=%d AND hostgroup_id IN ("
+				" SELECT %d UNION ALL"
+				" SELECT backup_writer_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d UNION ALL"
+				" SELECT offline_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d"
+			")";
+			sprintf(query,q,_hostname,_port,_writer_hostgroup,_writer_hostgroup,_writer_hostgroup);
 			mydb->execute(query);
 			//free(query);
-			q=(char *)"UPDATE mysql_servers_incoming SET status=0 WHERE hostname='%s' AND port=%d AND hostgroup_id=(SELECT reader_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d)";
-			//query=(char *)malloc(strlen(q)+strlen(_hostname)+64);
-			sprintf(query,q,_hostname,_port,_writer_hostgroup);
+			// NOTE: In case of the server being 'OFFLINE_SOFT' we preserve this status. Otherwise we set the server as 'ONLINE'.
+			q=(char *)"UPDATE mysql_servers_incoming SET status=(CASE "
+				" (SELECT status FROM mysql_servers_incoming WHERE hostname='%s' AND port=%d AND"
+					" hostgroup_id=(SELECT reader_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d)) WHEN 2 THEN 2 ELSE 0 END)"
+				" WHERE hostname='%s' AND port=%d AND hostgroup_id=(SELECT reader_hostgroup FROM mysql_group_replication_hostgroups WHERE writer_hostgroup=%d)";
+			sprintf(query,q,_hostname,_port,_writer_hostgroup,_hostname,_port,_writer_hostgroup);
 			mydb->execute(query);
 			//free(query);
 			converge_group_replication_config(_writer_hostgroup);
@@ -4680,7 +4812,12 @@ void MySQL_HostGroups_Manager::update_group_replication_set_writer(char *_hostna
 	bool found_writer=false;
 	bool found_reader=false;
 	int read_HG=-1;
+	int offline_HG=-1;
+	int backup_writer_HG=-1;
 	bool need_converge=false;
+	int status=0;
+	bool offline_soft_found=false;
+
 	if (resultset) {
 		// let's get info about this cluster
 		pthread_mutex_lock(&Group_Replication_Info_mutex);
@@ -4691,6 +4828,8 @@ void MySQL_HostGroups_Manager::update_group_replication_set_writer(char *_hostna
 			info=it2->second;
 			writer_is_also_reader=info->writer_is_also_reader;
 			read_HG=info->reader_hostgroup;
+			offline_HG=info->offline_hostgroup;
+			backup_writer_HG=info->backup_writer_hostgroup;
 			need_converge=info->need_converge;
 			info->need_converge=false;
 		}
@@ -4700,9 +4839,11 @@ void MySQL_HostGroups_Manager::update_group_replication_set_writer(char *_hostna
 			for (std::vector<SQLite3_row *>::iterator it = resultset->rows.begin() ; it != resultset->rows.end(); ++it) {
 				SQLite3_row *r=*it;
 				int hostgroup=atoi(r->fields[0]);
+				offline_soft_found = atoi(r->fields[1]) == 2 ? true : false;
+
 				if (hostgroup==_writer_hostgroup) {
-					int status = atoi(r->fields[1]);
-					if (status == 0) {
+					status = atoi(r->fields[1]);
+					if (status == 0 || status == 2) {
 						found_writer=true;
 					}
 				}
@@ -4713,6 +4854,13 @@ void MySQL_HostGroups_Manager::update_group_replication_set_writer(char *_hostna
 				}
 			}
 		}
+		// NOTE: In case of a writer not being found but a 'OFFLINE_SOFT' status
+		// is found in a hostgroup, 'OFFLINE_SOFT' status should be preserved.
+		if (found_writer == false) {
+			if (offline_soft_found) {
+				status = 2;
+			}
+		}
 		if (need_converge==false) {
 			if (found_writer) { // maybe no-op
 				if (
@@ -4735,19 +4883,19 @@ void MySQL_HostGroups_Manager::update_group_replication_set_writer(char *_hostna
 			GloAdmin->mysql_servers_wrlock();
 			mydb->execute("DELETE FROM mysql_servers_incoming");
 			mydb->execute("INSERT INTO mysql_servers_incoming SELECT hostgroup_id, hostname, port, gtid_port, weight, status, compression, max_connections, max_replication_lag, use_ssl, max_latency_ms, comment FROM mysql_servers");
-			q=(char *)"UPDATE OR IGNORE mysql_servers_incoming SET hostgroup_id=%d WHERE hostname='%s' AND port=%d AND hostgroup_id<>%d";
+			// NOTE: Only updated the servers that have belong to the same cluster.
+			q=(char *)"UPDATE OR IGNORE mysql_servers_incoming SET hostgroup_id=%d WHERE hostname='%s' AND port=%d AND hostgroup_id IN (%d, %d, %d)";
 			query=(char *)malloc(strlen(q)+strlen(_hostname)+256);
-			sprintf(query,q,_writer_hostgroup,_hostname,_port,_writer_hostgroup);
+			sprintf(query,q,_writer_hostgroup,_hostname,_port,backup_writer_HG,read_HG,offline_HG);
 			mydb->execute(query);
-			//free(query);
-			q=(char *)"DELETE FROM mysql_servers_incoming WHERE hostname='%s' AND port=%d AND hostgroup_id<>%d";
-			//query=(char *)malloc(strlen(q)+strlen(_hostname)+64);
-			sprintf(query,q,_hostname,_port,_writer_hostgroup);
+			// NOTE: Only delete the servers that have belong to the same cluster.
+			q=(char *)"DELETE FROM mysql_servers_incoming WHERE hostname='%s' AND port=%d AND hostgroup_id IN (%d, %d, %d)";
+			sprintf(query,q,_hostname,_port,backup_writer_HG,read_HG,offline_HG);
 			mydb->execute(query);
-			//free(query);
-			q=(char *)"UPDATE mysql_servers_incoming SET status=0 WHERE hostname='%s' AND port=%d AND hostgroup_id=%d";
-			//query=(char *)malloc(strlen(q)+strlen(_hostname)+64);
-			sprintf(query,q,_hostname,_port,_writer_hostgroup);
+			q=(char *)"UPDATE mysql_servers_incoming SET status=%d WHERE hostname='%s' AND port=%d AND hostgroup_id=%d";
+			// NOTE: In case of the server being 'OFFLINE_SOFT' we preserve this status. Otherwise
+			// we set the server as 'ONLINE'.
+			sprintf(query, q, (status == 2 ? 2 : 0 ), _hostname, _port, _writer_hostgroup);
 			mydb->execute(query);
 			//free(query);
 			if (writer_is_also_reader && read_HG>=0) {
diff --git a/lib/MySQL_Monitor.cpp b/lib/MySQL_Monitor.cpp
index fb7e6eed44..01517782c2 100644
--- a/lib/MySQL_Monitor.cpp
+++ b/lib/MySQL_Monitor.cpp
@@ -1420,7 +1420,11 @@ void * monitor_group_replication_thread(void *arg) {
 	//mmsd->async_exit_status=mysql_ping_start(&mmsd->interr,mmsd->mysql);
 	mmsd->interr=0; // reset the value
 #ifdef TEST_GROUPREP
-	mmsd->async_exit_status=mysql_query_start(&mmsd->interr,mmsd->mysql,"SELECT viable_candidate,read_only,transactions_behind FROM GR_MEMBER_ROUTING_CANDIDATE_STATUS");
+	{
+		std::string s { "SELECT viable_candidate,read_only,transactions_behind FROM GR_MEMBER_ROUTING_CANDIDATE_STATUS" };
+		s += " " + std::string(mmsd->hostname) + ":" + std::to_string(mmsd->port);
+		mmsd->async_exit_status=mysql_query_start(&mmsd->interr,mmsd->mysql,s.c_str());
+	}
 #else
 	mmsd->async_exit_status=mysql_query_start(&mmsd->interr,mmsd->mysql,"SELECT viable_candidate,read_only,transactions_behind FROM sys.gr_member_routing_candidate_status");
 #endif
@@ -1561,10 +1565,12 @@ void * monitor_group_replication_thread(void *arg) {
 					mmsd->hostname, mmsd->port, num_timeouts, mmsd->max_transactions_behind_count);
 			}
 		}
-		int lag_counts = 0;
-		if (read_only) {
-			lag_counts = node->get_lag_behind_count(mmsd->max_transactions_behind);
-		}
+		// NOTE: Previously 'lag_counts' was only updated for 'read_only'
+		// because 'writers' were never selected for being set 'OFFLINE' due to
+		// replication lag. Since the change of this behavior to 'SHUNNING'
+		// with replication lag, no matter it's 'read_only' value, 'lag_counts'
+		// is computed everytime.
+		int lag_counts = node->get_lag_behind_count(mmsd->max_transactions_behind);
 		pthread_mutex_unlock(&GloMyMon->group_replication_mutex);
 
 		// NOTE: we update MyHGM outside the mutex group_replication_mutex
@@ -1588,16 +1594,26 @@ void * monitor_group_replication_thread(void *arg) {
 				MyHGM->update_group_replication_set_offline(mmsd->hostname, mmsd->port, mmsd->writer_hostgroup, (char *)"viable_candidate=NO");
 			} else {
 				if (read_only==true) {
-					if (lag_counts >= mysql_thread___monitor_groupreplication_max_transactions_behind_count) {
-						MyHGM->update_group_replication_set_offline(mmsd->hostname, mmsd->port, mmsd->writer_hostgroup, (char *)"slave is lagging");
-					} else {
-						MyHGM->update_group_replication_set_read_only(mmsd->hostname, mmsd->port, mmsd->writer_hostgroup, (char *)"read_only=YES");
-					}
+					MyHGM->update_group_replication_set_read_only(mmsd->hostname, mmsd->port, mmsd->writer_hostgroup, (char *)"read_only=YES");
 				} else {
 					// the node is a writer
 					// TODO: for now we don't care about the number of writers
 					MyHGM->update_group_replication_set_writer(mmsd->hostname, mmsd->port, mmsd->writer_hostgroup);
 				}
+
+				// NOTE: Replication lag action should takes place **after** the
+				// servers have been placed in the correct hostgroups, otherwise
+				// during the reconfiguration of the servers due to 'update_group_replication_set_writer'
+				// there would be a small window in which the 'SHUNNED' server
+				// will be treat as 'ONLINE' letting some new connections to
+				// take places, before it becomes 'SHUNNED' again.
+				bool enable = true;
+				if (lag_counts >= mysql_thread___monitor_groupreplication_max_transactions_behind_count) {
+					enable = false;
+				}
+				MyHGM->group_replication_lag_action(
+					mmsd->writer_hostgroup, mmsd->hostname, mmsd->port, lag_counts, read_only, enable
+				);
 			}
 		}
 
diff --git a/lib/MySQL_Protocol.cpp b/lib/MySQL_Protocol.cpp
index e8ca27c0f8..c423a5efda 100644
--- a/lib/MySQL_Protocol.cpp
+++ b/lib/MySQL_Protocol.cpp
@@ -2233,7 +2233,12 @@ stmt_execute_metadata_t * MySQL_Protocol::get_binds_from_pkt(void *ptr, unsigned
 			uint8_t tmp_is_null = (null_byte & ( 1 << idx )) >> idx;
 			my_bool is_null = tmp_is_null;
 			if (new_params_bound_flag == 0) {
-				// TODO: Add comment
+				// NOTE: Just impose 'is_null' to be '1' using the values from
+				// previous bindings when we know values for these **haven't
+				// changed**, this is, when 'new_params_bound_flag' is '0'.
+				// Otherwise we will assume a value to be 'NULL' when the
+				// binding type could have actually been changed from the
+				// previous 'MYSQL_TYPE_NULL'. For more context see #3603.
 				if (binds[i].buffer_type == MYSQL_TYPE_NULL)
 					is_null = 1;
 			}
@@ -2265,7 +2270,7 @@ stmt_execute_metadata_t * MySQL_Protocol::get_binds_from_pkt(void *ptr, unsigned
 				}
 				binds[i].buffer_type=(enum enum_field_types)buffer_type;
 				// NOTE: This is required because further check for nullity rely on
-				// 'is_nulls' instead of 'buffer_type'.
+				// 'is_nulls' instead of 'buffer_type'. See #3603.
 				if (binds[i].buffer_type == MYSQL_TYPE_NULL) {
 					is_nulls[i]= 1;
 				}
diff --git a/lib/MySQL_Session.cpp b/lib/MySQL_Session.cpp
index ca7f62f436..82a14fb1c4 100644
--- a/lib/MySQL_Session.cpp
+++ b/lib/MySQL_Session.cpp
@@ -4816,6 +4816,7 @@ void MySQL_Session::handler___status_CHANGING_USER_CLIENT___STATE_CLIENT_HANDSHA
 void MySQL_Session::handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(PtrSize_t *pkt, bool *wrong_pass) {
 	bool is_encrypted = client_myds->encrypted;
 	bool handshake_response_return = client_myds->myprot.process_pkt_handshake_response((unsigned char *)pkt->ptr,pkt->size);
+	bool handshake_err = true;
 
 	proxy_debug(PROXY_DEBUG_MYSQL_CONNECTION,8,"Session=%p , DS=%p , handshake_response=%d , switching_auth_stage=%d , is_encrypted=%d , client_encrypted=%d\n", this, client_myds, handshake_response_return, client_myds->switching_auth_stage, is_encrypted, client_myds->encrypted);
 	if (
@@ -4840,6 +4841,7 @@ void MySQL_Session::handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(
 		SSL_set_fd(client_myds->ssl, client_myds->fd);
 		SSL_set_accept_state(client_myds->ssl); 
 		SSL_set_bio(client_myds->ssl, client_myds->rbio_ssl, client_myds->wbio_ssl);
+		l_free(pkt->size,pkt->ptr);
 		return;
 	}
 
@@ -5001,6 +5003,7 @@ void MySQL_Session::handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(
 					) {
 						// we are good!
 						client_myds->myprot.generate_pkt_OK(true,NULL,NULL, _pid, 0,0,0,0,NULL);
+						handshake_err = false;
 						GloMyLogger->log_audit_entry(PROXYSQL_MYSQL_AUTH_OK, this, NULL);
 						status=WAITING_CLIENT_DATA;
 						client_myds->DSS=STATE_CLIENT_AUTH_OK;
@@ -5037,6 +5040,7 @@ void MySQL_Session::handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(
 						proxy_debug(PROXY_DEBUG_MYSQL_CONNECTION,8,"Session=%p , DS=%p . STATE_CLIENT_AUTH_OK\n", this, client_myds);
 						GloMyLogger->log_audit_entry(PROXYSQL_MYSQL_AUTH_OK, this, NULL);
 						client_myds->myprot.generate_pkt_OK(true,NULL,NULL, _pid, 0,0,0,0,NULL);
+						handshake_err = false;
 						status=WAITING_CLIENT_DATA;
 						client_myds->DSS=STATE_CLIENT_AUTH_OK;
 					}
@@ -5099,10 +5103,17 @@ void MySQL_Session::handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(
 			free(_s);
 			__sync_fetch_and_add(&MyHGM->status.access_denied_wrong_password, 1);
 		}
+		if (client_addr) {
+			free(client_addr);
+		}
 		GloMyLogger->log_audit_entry(PROXYSQL_MYSQL_AUTH_ERR, this, NULL);
 		__sync_add_and_fetch(&MyHGM->status.client_connections_aborted,1);
 		client_myds->DSS=STATE_SLEEP;
 	}
+
+	if (mysql_thread___client_host_cache_size) {
+		GloMTH->update_client_host_cache(client_myds->client_addr, handshake_err);
+	}
 }
 
 // Note: as commented in issue #546 and #547 , some clients ignore the status of CLIENT_MULTI_STATEMENTS
diff --git a/lib/MySQL_Thread.cpp b/lib/MySQL_Thread.cpp
index 4c406d10db..4c509a1396 100644
--- a/lib/MySQL_Thread.cpp
+++ b/lib/MySQL_Thread.cpp
@@ -88,6 +88,7 @@ mythr_st_vars_t MySQL_Thread_status_variables_counter_array[] {
 	{ st_var_whitelisted_sqli_fingerprint,p_th_counter::whitelisted_sqli_fingerprint,     (char *)"whitelisted_sqli_fingerprint" },
 	{ st_var_max_connect_timeout_err,     p_th_counter::max_connect_timeouts,             (char *)"max_connect_timeouts" },
 	{ st_var_generated_pkt_err,           p_th_counter::generated_error_packets,          (char *)"generated_error_packets" },
+	{ st_var_client_host_error_killed_connections, p_th_counter::client_host_error_killed_connections, (char *)"client_host_error_killed_connections" },
 };
 
 mythr_g_st_vars_t MySQL_Thread_status_variables_gauge_array[] {
@@ -412,6 +413,8 @@ static char * mysql_thread_variables_names[]= {
 	(char *)"shun_recovery_time_sec",
 	(char *)"query_retries_on_failure",
 	(char *)"client_multi_statements",
+	(char *)"client_host_cache_size",
+	(char *)"client_host_error_counts",
 	(char *)"connect_retries_on_failure",
 	(char *)"connect_retries_delay",
 	(char *)"connection_delay_multiplex_ms",
@@ -457,6 +460,7 @@ static char * mysql_thread_variables_names[]= {
 	(char *)"monitor_groupreplication_healthcheck_timeout",
 	(char *)"monitor_groupreplication_healthcheck_max_timeout_count",
 	(char *)"monitor_groupreplication_max_transactions_behind_count",
+	(char *)"monitor_groupreplication_max_transactions_behind_for_read_only",
 	(char *)"monitor_galera_healthcheck_interval",
 	(char *)"monitor_galera_healthcheck_timeout",
 	(char *)"monitor_galera_healthcheck_max_timeout_count",
@@ -866,6 +870,12 @@ th_metrics_map = std::make_tuple(
 			"proxysql_mysql_killed_backend_queries_total",
 			"Killed backend queries.",
 			metric_tags {}
+		),
+		std::make_tuple (
+			p_th_counter::client_host_error_killed_connections,
+			"proxysql_client_host_error_killed_connections",
+			"Killed client connections because address exceeded 'client_host_error_counts'.",
+			metric_tags {}
 		)
 	},
 	th_gauge_vector {
@@ -1019,10 +1029,14 @@ MySQL_Threads_Handler::MySQL_Threads_Handler() {
 	shutdown_=0;
 	pthread_rwlock_init(&rwlock,NULL);
 	pthread_attr_init(&attr);
+	// Zero initialize all variables
+	memset(&variables, 0, sizeof(variables));
 	variables.shun_on_failures=5;
 	variables.shun_recovery_time_sec=10;
 	variables.query_retries_on_failure=1;
 	variables.client_multi_statements=true;
+	variables.client_host_cache_size=0;
+	variables.client_host_error_counts=0;
 	variables.connect_retries_on_failure=10;
 	variables.connection_delay_multiplex_ms=0;
 	variables.connection_max_age_ms=0;
@@ -1048,6 +1062,7 @@ MySQL_Threads_Handler::MySQL_Threads_Handler() {
 	variables.monitor_groupreplication_healthcheck_timeout=800;
 	variables.monitor_groupreplication_healthcheck_max_timeout_count=3;
 	variables.monitor_groupreplication_max_transactions_behind_count=3;
+	variables.monitor_groupreplication_max_transactions_behind_for_read_only=1;
 	variables.monitor_galera_healthcheck_interval=5000;
 	variables.monitor_galera_healthcheck_timeout=800;
 	variables.monitor_galera_healthcheck_max_timeout_count=3;
@@ -1177,8 +1192,10 @@ MySQL_Threads_Handler::MySQL_Threads_Handler() {
 	// Initialize prometheus metrics
 	init_prometheus_counter_array<th_metrics_map_idx, p_th_counter>(th_metrics_map, this->status_variables.p_counter_array);
 	init_prometheus_gauge_array<th_metrics_map_idx, p_th_gauge>(th_metrics_map, this->status_variables.p_gauge_array);
-}
 
+	// Init client_host_cache mutex
+	pthread_mutex_init(&mutex_client_host_cache, NULL);
+}
 
 unsigned int MySQL_Threads_Handler::get_global_version() {
 	return __sync_fetch_and_add(&__global_MySQL_Thread_Variables_version,0);
@@ -2116,6 +2133,7 @@ char ** MySQL_Threads_Handler::get_variables_list() {
 		VariablesPointers_int["monitor_groupreplication_healthcheck_timeout"]           = make_tuple(&variables.monitor_groupreplication_healthcheck_timeout,           100,       600*1000, false);
 		VariablesPointers_int["monitor_groupreplication_healthcheck_max_timeout_count"] = make_tuple(&variables.monitor_groupreplication_healthcheck_max_timeout_count,   1,             10, false);
 		VariablesPointers_int["monitor_groupreplication_max_transactions_behind_count"] = make_tuple(&variables.monitor_groupreplication_max_transactions_behind_count,   1,             10, false);
+		VariablesPointers_int["monitor_groupreplication_max_transactions_behind_for_read_only"] = make_tuple(&variables.monitor_groupreplication_max_transactions_behind_for_read_only,   0, 2, false);
 
 		VariablesPointers_int["monitor_galera_healthcheck_interval"]          = make_tuple(&variables.monitor_galera_healthcheck_interval,          50, 7*24*3600*1000, false);
 		VariablesPointers_int["monitor_galera_healthcheck_timeout"]           = make_tuple(&variables.monitor_galera_healthcheck_timeout,           50,       600*1000, false);
@@ -2173,6 +2191,9 @@ char ** MySQL_Threads_Handler::get_variables_list() {
 		VariablesPointers_int["handle_unknown_charset"]        = make_tuple(&variables.handle_unknown_charset,        0, HANDLE_UNKNOWN_CHARSET__MAX_HANDLE_VALUE, false);
 		VariablesPointers_int["ping_interval_server_msec"]     = make_tuple(&variables.ping_interval_server_msec,  1000, 7*24*3600*1000, false);
 		VariablesPointers_int["ping_timeout_server"]           = make_tuple(&variables.ping_timeout_server,          10,       600*1000, false);
+		VariablesPointers_int["client_host_cache_size"]        = make_tuple(&variables.client_host_cache_size,        0,      1024*1024, false);
+		VariablesPointers_int["client_host_error_counts"]      = make_tuple(&variables.client_host_error_counts,      0,      1024*1024, false);
+
 		// logs
 		VariablesPointers_int["auditlog_filesize"]     = make_tuple(&variables.auditlog_filesize,    1024*1024, 1*1024*1024*1024, false);
 		VariablesPointers_int["eventslog_filesize"]    = make_tuple(&variables.eventslog_filesize,   1024*1024, 1*1024*1024*1024, false);
@@ -2372,6 +2393,204 @@ void MySQL_Threads_Handler::stop_listeners() {
 	free_tokenizer( &tok );
 }
 
+/**
+ * @brief Gets the client address stored in 'client_addr' member as
+ *   an string if available. If member 'client_addr' is NULL, returns an
+ *   empty string.
+ *
+ * @return Either an string holding the string representation of internal
+ *   member 'client_addr', or empty string if this member is NULL.
+ */
+std::string get_client_addr(struct sockaddr* client_addr) {
+	char buf[INET6_ADDRSTRLEN];
+	std::string str_client_addr {};
+
+	if (client_addr == NULL) {
+		return str_client_addr;
+	}
+
+	switch (client_addr->sa_family) {
+		case AF_INET: {
+			struct sockaddr_in *ipv4 = (struct sockaddr_in *)client_addr;
+			inet_ntop(client_addr->sa_family, &ipv4->sin_addr, buf, INET_ADDRSTRLEN);
+			str_client_addr = std::string { buf };
+			break;
+		}
+		case AF_INET6: {
+			struct sockaddr_in6 *ipv6 = (struct sockaddr_in6 *)client_addr;
+			inet_ntop(client_addr->sa_family, &ipv6->sin6_addr, buf, INET6_ADDRSTRLEN);
+			str_client_addr = std::string { buf };
+			break;
+		}
+		default:
+			str_client_addr = std::string { "localhost" };
+			break;
+	}
+
+	return str_client_addr;
+}
+
+MySQL_Client_Host_Cache_Entry MySQL_Threads_Handler::find_client_host_cache(struct sockaddr* client_sockaddr) {
+	MySQL_Client_Host_Cache_Entry entry { 0, 0 };
+	if (client_sockaddr == NULL) {
+		// NOTE: client_sockaddr **shouldn't** ever by 'NULL', no matter the
+		// 'session_type' in from which this function is called. Because
+		// `MySQL_Session::client_myds::client_addr` should **always** be
+		// initialized before `handler` is called.
+		assert(0);
+	}
+	if (client_sockaddr->sa_family != AF_INET && client_sockaddr->sa_family != AF_INET6) {
+		return entry;
+	}
+	std::string client_addr = get_client_addr(client_sockaddr);
+	if (client_addr == "127.0.0.1") {
+		return entry;
+	}
+
+	pthread_mutex_lock(&mutex_client_host_cache);
+	auto found_entry = client_host_cache.find(client_addr);
+	if (found_entry != client_host_cache.end()) {
+		entry = found_entry->second;
+	}
+	pthread_mutex_unlock(&mutex_client_host_cache);
+
+	return entry;
+}
+
+/**
+ * @brief Number of columns for representing a 'MySQL_Client_Host_Cache_Entry'
+ *   in a 'SQLite3_result'.
+ */
+const int CLIENT_HOST_CACHE_COLUMNS = 3;
+
+/**
+ * @brief Helper function that converts a given client address and a
+ *   'MySQL_Client_Host_Cache_Entry', into a row for a 'SQLite3_result' for
+ *   table 'STATS_SQLITE_TABLE_MYSQL_CLIENT_HOST_CACHE'.
+ *
+ * @param address The client address to be added to the resulset row.
+ * @param entry The 'MySQL_Client_Host_Cache_Entry' to be added to the resulset
+ *   row.
+ *
+ * @return A pointer array holding the values for each of the columns of the
+ *   row. It should be freed through helper function 'free_client_host_cache_row'.
+ */
+char** client_host_cache_entry_row(
+	const std::string address, const MySQL_Client_Host_Cache_Entry& entry
+) {
+	// INET6_ADDRSTRLEN length should be enough for holding any member:
+	//  { address: MAX INET6_ADDRSTRLEN, updated_at: uint64_t, error_count: uint32_t }
+	char buff[INET6_ADDRSTRLEN];
+	char** row =
+		static_cast<char**>(malloc(sizeof(char*)*CLIENT_HOST_CACHE_COLUMNS));
+
+	row[0]=strdup(address.c_str());
+	sprintf(buff, "%u", entry.error_count);
+	row[1]=strdup(buff);
+	sprintf(buff, "%lu", entry.updated_at);
+	row[2]=strdup(buff);
+
+	return row;
+}
+
+/**
+ * @brief Helper function to free the row returned by
+ * 'client_host_cache_entry_row'.
+ *
+ * @param row The pointer array holding the row values to be freed.
+ */
+void free_client_host_cache_row(char** row) {
+	for (int i = 0; i < CLIENT_HOST_CACHE_COLUMNS; i++) {
+		free(row[i]);
+	}
+	free(row);
+}
+
+SQLite3_result* MySQL_Threads_Handler::get_client_host_cache(bool reset) {
+	SQLite3_result *result = new SQLite3_result(CLIENT_HOST_CACHE_COLUMNS);
+
+	pthread_mutex_lock(&mutex_client_host_cache);
+	result->add_column_definition(SQLITE_TEXT,"client_address");
+	result->add_column_definition(SQLITE_TEXT,"error_count");
+	result->add_column_definition(SQLITE_TEXT,"last_updated");
+
+	for (const auto& cache_entry : client_host_cache) {
+		char** row = client_host_cache_entry_row(cache_entry.first, cache_entry.second);
+		result->add_row(row);
+		free_client_host_cache_row(row);
+	}
+
+	if (reset) {
+		client_host_cache.clear();
+	}
+
+	pthread_mutex_unlock(&mutex_client_host_cache);
+	return result;
+}
+
+void MySQL_Threads_Handler::update_client_host_cache(struct sockaddr* client_sockaddr, bool error) {
+	if (client_sockaddr == NULL) {
+		// NOTE: client_sockaddr **shouldn't** ever by 'NULL', no matter the
+		// 'session_type' in from which this function is called. Because
+		// `MySQL_Session::client_myds::client_addr` should **always** be
+		// initialized before `handler` is called.
+		assert(0);
+	}
+	if (client_sockaddr->sa_family != AF_INET && client_sockaddr->sa_family != AF_INET6) {
+		return;
+	}
+	std::string client_addr = get_client_addr(client_sockaddr);
+	if (client_addr == "127.0.0.1") {
+		return;
+	}
+
+	if (error) {
+		pthread_mutex_lock(&mutex_client_host_cache);
+		// If the cache is full, find the oldest entry on it, and update/remove it.
+		if (client_host_cache.size() >= static_cast<size_t>(mysql_thread___client_host_cache_size)) {
+			auto older_elem = std::min_element(
+				client_host_cache.begin(),
+				client_host_cache.end(),
+				[] (const std::pair<std::string, MySQL_Client_Host_Cache_Entry>& f_entry,
+					const std::pair<std::string, MySQL_Client_Host_Cache_Entry>& s_entry)
+				{
+					return f_entry.second.updated_at < s_entry.second.updated_at;
+				}
+			);
+			if (older_elem->first != client_addr) {
+				client_host_cache.erase(older_elem);
+			}
+		}
+
+		// Find the entry for the client, and update/insert it.
+		auto cache_entry = client_host_cache.find(client_addr);
+		if (cache_entry != client_host_cache.end()) {
+			cache_entry->second.error_count += 1;
+			cache_entry->second.updated_at = monotonic_time();
+		} else {
+			// Notice than the value of 'mysql_thread___client_host_cache_size' can
+			// change at runtime. Due to this, we should only insert when the size of the
+			// cache is smaller than this value, otherwise we could end in situations in
+			// which cache doesn't shrink after it's size is reduced at runtime.
+			if (client_host_cache.size() < static_cast<size_t>(mysql_thread___client_host_cache_size)) {
+				MySQL_Client_Host_Cache_Entry new_entry { monotonic_time(), 1 };
+				client_host_cache.insert({client_addr, new_entry});
+			}
+		}
+		pthread_mutex_unlock(&mutex_client_host_cache);
+	} else {
+		pthread_mutex_lock(&mutex_client_host_cache);
+		client_host_cache.erase(client_addr);
+		pthread_mutex_unlock(&mutex_client_host_cache);
+	}
+}
+
+void MySQL_Threads_Handler::flush_client_host_cache() {
+	pthread_mutex_lock(&mutex_client_host_cache);
+	client_host_cache.clear();
+	pthread_mutex_unlock(&mutex_client_host_cache);
+}
+
 MySQL_Threads_Handler::~MySQL_Threads_Handler() {
 	if (variables.monitor_username) { free(variables.monitor_username); variables.monitor_username=NULL; }
 	if (variables.monitor_password) { free(variables.monitor_password); variables.monitor_password=NULL; }
@@ -3496,6 +3715,7 @@ void MySQL_Thread::process_all_sessions() {
 			if (sess_time/1000 > (unsigned long long)mysql_thread___connect_timeout_client) {
 				proxy_warning("Closing not established client connection %s:%d after %llums\n",sess->client_myds->addr.addr,sess->client_myds->addr.port, sess_time/1000);
 				sess->healthy = 0;
+				GloMTH->update_client_host_cache(sess->client_myds->client_addr, true);
 			}
 		}
 		if (maintenance_loop) {
@@ -3669,6 +3889,7 @@ void MySQL_Thread::refresh_variables() {
 	mysql_thread___monitor_groupreplication_healthcheck_timeout=GloMTH->get_variable_int((char *)"monitor_groupreplication_healthcheck_timeout");
 	mysql_thread___monitor_groupreplication_healthcheck_max_timeout_count=GloMTH->get_variable_int((char *)"monitor_groupreplication_healthcheck_max_timeout_count");
 	mysql_thread___monitor_groupreplication_max_transactions_behind_count=GloMTH->get_variable_int((char *)"monitor_groupreplication_max_transactions_behind_count");
+	mysql_thread___monitor_groupreplication_max_transaction_behind_for_read_only=GloMTH->get_variable_int((char *)"monitor_groupreplication_max_transactions_behind_for_read_only");
 	mysql_thread___monitor_galera_healthcheck_interval=GloMTH->get_variable_int((char *)"monitor_galera_healthcheck_interval");
 	mysql_thread___monitor_galera_healthcheck_timeout=GloMTH->get_variable_int((char *)"monitor_galera_healthcheck_timeout");
 	mysql_thread___monitor_galera_healthcheck_max_timeout_count=GloMTH->get_variable_int((char *)"monitor_galera_healthcheck_max_timeout_count");
@@ -3761,6 +3982,8 @@ void MySQL_Thread::refresh_variables() {
 	mysql_thread___enable_server_deprecate_eof=(bool)GloMTH->get_variable_int((char *)"enable_server_deprecate_eof");
 	mysql_thread___enable_load_data_local_infile=(bool)GloMTH->get_variable_int((char *)"enable_load_data_local_infile");
 	mysql_thread___log_mysql_warnings_enabled=(bool)GloMTH->get_variable_int((char *)"log_mysql_warnings_enabled");
+	mysql_thread___client_host_cache_size=GloMTH->get_variable_int((char *)"client_host_cache_size");
+	mysql_thread___client_host_error_counts=GloMTH->get_variable_int((char *)"client_host_error_counts");
 #ifdef DEBUG
 	mysql_thread___session_debug=(bool)GloMTH->get_variable_int((char *)"session_debug");
 #endif /* DEBUG */
@@ -3840,7 +4063,6 @@ void MySQL_Thread::unregister_session_connection_handler(int idx, bool _new) {
 	mysql_sessions->remove_index_fast(idx);
 }
 
-
 void MySQL_Thread::listener_handle_new_connection(MySQL_Data_Stream *myds, unsigned int n) {
 	int c;
 	union {
@@ -3862,6 +4084,25 @@ void MySQL_Thread::listener_handle_new_connection(MySQL_Data_Stream *myds, unsig
 	}
 	c=accept(myds->fd, addr, &addrlen);
 	if (c>-1) { // accept() succeeded
+		if (mysql_thread___client_host_cache_size) {
+			MySQL_Client_Host_Cache_Entry client_host_entry =
+				GloMTH->find_client_host_cache(addr);
+			if (
+				client_host_entry.updated_at != 0 &&
+				client_host_entry.error_count >= static_cast<uint32_t>(mysql_thread___client_host_error_counts)
+			) {
+				std::string client_addr = get_client_addr(addr);
+				proxy_error(
+					"Closing connection because client '%s' reached 'mysql-client_host_error_counts': %d\n",
+					client_addr.c_str(), mysql_thread___client_host_error_counts
+				);
+				close(c);
+				free(addr);
+				status_variables.stvar[st_var_client_host_error_killed_connections] += 1;
+				return;
+			}
+		}
+
 		// create a new client connection
 		mypolls.fds[n].revents=0;
 		MySQL_Session *sess=create_new_session_and_client_data_stream(c);
diff --git a/lib/ProxySQL_Admin.cpp b/lib/ProxySQL_Admin.cpp
index 4f4690fcdb..219cc07974 100644
--- a/lib/ProxySQL_Admin.cpp
+++ b/lib/ProxySQL_Admin.cpp
@@ -490,6 +490,9 @@ static int http_handler(void *cls, struct MHD_Connection *connection, const char
 #define STATS_SQLITE_TABLE_MYSQL_ERRORS "CREATE TABLE stats_mysql_errors (hostgroup INT NOT NULL , hostname VARCHAR NOT NULL , port INT NOT NULL , username VARCHAR NOT NULL , client_address VARCHAR NOT NULL , schemaname VARCHAR NOT NULL , errno INT NOT NULL , count_star INTEGER NOT NULL , first_seen INTEGER NOT NULL , last_seen INTEGER NOT NULL , last_error VARCHAR NOT NULL DEFAULT '' , PRIMARY KEY (hostgroup, hostname, port, username, schemaname, errno) )"
 #define STATS_SQLITE_TABLE_MYSQL_ERRORS_RESET "CREATE TABLE stats_mysql_errors_reset (hostgroup INT NOT NULL , hostname VARCHAR NOT NULL , port INT NOT NULL , username VARCHAR NOT NULL , client_address VARCHAR NOT NULL , schemaname VARCHAR NOT NULL , errno INT NOT NULL , count_star INTEGER NOT NULL , first_seen INTEGER NOT NULL , last_seen INTEGER NOT NULL , last_error VARCHAR NOT NULL DEFAULT '' , PRIMARY KEY (hostgroup, hostname, port, username, schemaname, errno) )"
 
+#define STATS_SQLITE_TABLE_MYSQL_CLIENT_HOST_CACHE "CREATE TABLE stats_mysql_client_host_cache (client_address VARCHAR NOT NULL, error_count INT NOT NULL, last_updated BIGINT NOT NULL)"
+#define STATS_SQLITE_TABLE_MYSQL_CLIENT_HOST_CACHE_RESET "CREATE TABLE stats_mysql_client_host_cache_reset (client_address VARCHAR NOT NULL, error_count INT NOT NULL, last_updated BIGINT NOT NULL)"
+
 #ifdef DEBUG
 #define ADMIN_SQLITE_TABLE_DEBUG_LEVELS "CREATE TABLE debug_levels (module VARCHAR NOT NULL PRIMARY KEY , verbosity INT NOT NULL DEFAULT 0)"
 #define ADMIN_SQLITE_TABLE_DEBUG_FILTERS "CREATE TABLE debug_filters (filename VARCHAR NOT NULL , line INT NOT NULL , funct VARCHAR NOT NULL , PRIMARY KEY (filename, line, funct) )"
@@ -1530,6 +1533,16 @@ bool admin_handler_command_proxysql(char *query_no_space, unsigned int query_no_
 		return false;
 	}
 
+	if (!strcasecmp("PROXYSQL FLUSH MYSQL CLIENT HOSTS", query_no_space)) {
+		proxy_info("Received PROXYSQL FLUSH MYSQL CLIENT HOSTS command\n");
+		ProxySQL_Admin *SPA=(ProxySQL_Admin *)pa;
+		if (GloMTH) {
+			GloMTH->flush_client_host_cache();
+		}
+		SPA->send_MySQL_OK(&sess->client_myds->myprot, NULL);
+		return false;
+	}
+
 	if (
 		(query_no_space_length==strlen("PROXYSQL FLUSH CONFIGDB") && !strncasecmp("PROXYSQL FLUSH CONFIGDB",query_no_space, query_no_space_length)) // see #923
 	) {
@@ -2925,6 +2938,8 @@ bool ProxySQL_Admin::GenericRefreshStatistics(const char *query_no_space, unsign
 	bool stats_mysql_query_rules=false;
 	bool stats_mysql_users=false;
 	bool stats_mysql_gtid_executed=false;
+	bool stats_mysql_client_host_cache=false;
+	bool stats_mysql_client_host_cache_reset=false;
 	bool dump_global_variables=false;
 
 	bool runtime_scheduler=false;
@@ -3012,6 +3027,10 @@ bool ProxySQL_Admin::GenericRefreshStatistics(const char *query_no_space, unsign
 		{ stats_mysql_users=true; refresh=true; }
 	if (strstr(query_no_space,"stats_mysql_gtid_executed"))
 		{ stats_mysql_gtid_executed=true; refresh=true; }
+	if (strstr(query_no_space,"stats_mysql_client_host_cache"))
+		{ stats_mysql_client_host_cache=true; refresh=true; }
+	if (strstr(query_no_space,"stats_mysql_client_host_cache_reset"))
+		{ stats_mysql_client_host_cache_reset=true; refresh=true; }
 
 	if (strstr(query_no_space,"stats_proxysql_servers_checksums"))
 		{ stats_proxysql_servers_checksums = true; refresh = true; }
@@ -3152,6 +3171,13 @@ bool ProxySQL_Admin::GenericRefreshStatistics(const char *query_no_space, unsign
 			stats___mysql_prepared_statements_info();
 		}
 
+		if (stats_mysql_client_host_cache) {
+			stats___mysql_client_host_cache(false);
+		}
+		if (stats_mysql_client_host_cache_reset) {
+			stats___mysql_client_host_cache(true);
+		}
+
 		if (admin) {
 			if (dump_global_variables) {
 				pthread_mutex_lock(&GloVars.checksum_mutex);
@@ -5526,6 +5552,8 @@ bool ProxySQL_Admin::init() {
 	insert_into_tables_defs(tables_defs_stats,"stats_mysql_users", STATS_SQLITE_TABLE_MYSQL_USERS);
 	insert_into_tables_defs(tables_defs_stats,"global_variables", ADMIN_SQLITE_TABLE_GLOBAL_VARIABLES); // workaround for issue #708
 	insert_into_tables_defs(tables_defs_stats,"stats_mysql_prepared_statements_info", ADMIN_SQLITE_TABLE_STATS_MYSQL_PREPARED_STATEMENTS_INFO);
+	insert_into_tables_defs(tables_defs_stats,"stats_mysql_client_host_cache", STATS_SQLITE_TABLE_MYSQL_CLIENT_HOST_CACHE);
+	insert_into_tables_defs(tables_defs_stats,"stats_mysql_client_host_cache_reset", STATS_SQLITE_TABLE_MYSQL_CLIENT_HOST_CACHE_RESET);
 
 	// ProxySQL Cluster
 	insert_into_tables_defs(tables_defs_admin,"proxysql_servers", ADMIN_SQLITE_TABLE_PROXYSQL_SERVERS);
@@ -8759,6 +8787,52 @@ void ProxySQL_Admin::stats___mysql_query_digests(bool reset, bool copy) {
 	delete resultset;
 }
 
+void ProxySQL_Admin::stats___mysql_client_host_cache(bool reset) {
+	if (!GloQPro) return;
+
+	SQLite3_result* resultset = GloMTH->get_client_host_cache(reset);
+	if (resultset==NULL) return;
+
+	statsdb->execute("BEGIN");
+
+	int rc = 0;
+	sqlite3_stmt* statement=NULL;
+	char* query = NULL;
+
+	if (reset) {
+		query=(char*)"INSERT INTO stats_mysql_client_host_cache_reset VALUES (?1, ?2, ?3)";
+	} else {
+		query=(char*)"INSERT INTO stats_mysql_client_host_cache VALUES (?1, ?2, ?3)";
+	}
+
+	statsdb->execute("DELETE FROM stats_mysql_client_host_cache_reset");
+	statsdb->execute("DELETE FROM stats_mysql_client_host_cache");
+
+	rc = statsdb->prepare_v2(query, &statement);
+	ASSERT_SQLITE_OK(rc, statsdb);
+
+	for (std::vector<SQLite3_row *>::iterator it = resultset->rows.begin() ; it != resultset->rows.end(); ++it) {
+		SQLite3_row *row = *it;
+
+		rc=(*proxy_sqlite3_bind_text)(statement, 1, row->fields[0], -1, SQLITE_TRANSIENT); ASSERT_SQLITE_OK(rc, statsdb);
+		rc=(*proxy_sqlite3_bind_int64)(statement, 2, atoll(row->fields[1])); ASSERT_SQLITE_OK(rc, statsdb);
+		rc=(*proxy_sqlite3_bind_int64)(statement, 3, atoll(row->fields[2])); ASSERT_SQLITE_OK(rc, statsdb);
+
+		SAFE_SQLITE3_STEP2(statement);
+		rc=(*proxy_sqlite3_clear_bindings)(statement);
+		rc=(*proxy_sqlite3_reset)(statement);
+	}
+
+	(*proxy_sqlite3_finalize)(statement);
+
+	if (reset) {
+		statsdb->execute("INSERT INTO stats_mysql_client_host_cache SELECT * FROM stats_mysql_client_host_cache_reset");
+	}
+
+	statsdb->execute("COMMIT");
+	delete resultset;
+}
+
 void ProxySQL_Admin::stats___mysql_errors(bool reset) {
 	if (!GloQPro) return;
 	SQLite3_result * resultset=NULL;
diff --git a/src/SQLite3_Server.cpp b/src/SQLite3_Server.cpp
index b793c21b59..d14d71afe3 100644
--- a/src/SQLite3_Server.cpp
+++ b/src/SQLite3_Server.cpp
@@ -8,6 +8,7 @@
 
 #include "MySQL_Logger.hpp"
 #include "MySQL_Data_Stream.h"
+#include "proxysql_utils.h"
 #include "query_processor.h"
 #include "SQLite3_Server.h"
 
@@ -233,7 +234,80 @@ class sqlite3server_main_loop_listeners {
 
 static sqlite3server_main_loop_listeners S_amll;
 
+/**
+ * @brief Helper function that checks if the supplied string
+ *   is a number.
+ * @param s The string to check.
+ * @return True if the supplied string is just composed of
+ *   digits, false otherwise.
+ */
+bool is_number(const std::string& s) {
+	if (s.empty()) { return false; }
+
+	for (const auto& d : s) {
+		if (std::isdigit(d) == false) {
+			return false;
+		}
+	}
+
+	return true;
+}
+
+/**
+ * @brief Checks if the query matches an specified 'monitor_query' of the
+ *   following format:
+ *
+ *   "$MONITOR_QUERY" + " hostname:port"
+ *
+ *   If the query matches, 'true' is returned, false otherwise.
+ *
+ * @param monitor_query Query that should be matched against the current
+ *   supplied 'query'.
+ * @param query Current query, to be matched against the supplied
+ *   'monitor_query'.
+ * @return 'true' if the query matches, false otherwise.
+ */
+bool match_monitor_query(const std::string& monitor_query, const std::string& query) {
+	if (query.rfind(monitor_query, 0) != 0) {
+		return false;
+	}
+
+	std::string srv_address {
+		query.substr(monitor_query.size())
+	};
+
+	// Check that what is beyond this point, is just the servers address,
+	// written as an identifier 'n.n.n.n:n'.
+	std::size_t cur_mark_pos = 0;
+	for (int i = 0; i < 3; i++) {
+		std::size_t next_mark_pos = srv_address.find('.', cur_mark_pos);
+		if (next_mark_pos == std::string::npos) {
+			return false;
+		} else {
+			std::string number {
+				srv_address.substr(cur_mark_pos, next_mark_pos - cur_mark_pos)
+			};
+
+			if (is_number(number)) {
+				cur_mark_pos = next_mark_pos + 1;
+			} else {
+				return false;
+			}
+		}
+	}
 
+	// Check last part is also a valid number
+	cur_mark_pos = srv_address.find(':', cur_mark_pos);
+	if (cur_mark_pos == std::string::npos) {
+		return false;
+	} else {
+		std::string number {
+			srv_address.substr(cur_mark_pos + 1)
+		};
+
+		return is_number(number);
+	}
+}
 
 void SQLite3_Server_session_handler(MySQL_Session *sess, void *_pa, PtrSize_t *pkt) {
 
@@ -555,8 +629,41 @@ void SQLite3_Server_session_handler(MySQL_Session *sess, void *_pa, PtrSize_t *p
 #ifdef TEST_GROUPREP
 			if (strstr(query_no_space,(char *)"GR_MEMBER_ROUTING_CANDIDATE_STATUS")) {
 				pthread_mutex_lock(&GloSQLite3Server->grouprep_mutex);
-				GloSQLite3Server->populate_grouprep_table(sess, testLag);
-				if (testLag > 0) testLag--;
+				GloSQLite3Server->populate_grouprep_table(sess, 0);
+				// NOTE: This query should be in one place that can be reused by
+				// 'ProxySQL_Monitor' module.
+				const std::string grouprep_monitor_test_query_start {
+					"SELECT viable_candidate,read_only,transactions_behind "
+						"FROM GR_MEMBER_ROUTING_CANDIDATE_STATUS "
+				};
+
+				// If the query matches 'grouprep_monitor_test_query_start', it
+				// means that the query has been issued by `ProxySQL_Monitor` and
+				// we need to fetch for the proper values and replace the query
+				// with one holding the values from `grouprep_map`.
+				if (match_monitor_query(grouprep_monitor_test_query_start, query_no_space)) {
+					std::string srv_addr {
+						query_no_space + grouprep_monitor_test_query_start.size()
+					};
+
+					const group_rep_status& gr_srv_status =
+						GloSQLite3Server->grouprep_test_value(srv_addr);
+					free(query);
+
+					std::string t_select_as_query {
+						"SELECT '%s' AS viable_candidate, '%s' AS read_only, %d AS transactions_behind"
+					};
+					std::string select_as_query {};
+					string_format(
+						t_select_as_query, select_as_query,
+						std::get<0>(gr_srv_status) ? "YES" : "NO",
+						std::get<1>(gr_srv_status) ? "YES" : "NO",
+						std::get<2>(gr_srv_status)
+					);
+
+					query = static_cast<char*>(malloc(select_as_query.length() + 1));
+					strcpy(query, select_as_query.c_str());
+				}
 			}
 #endif // TEST_GROUPREP
 			if (strstr(query_no_space,(char *)"Seconds_Behind_Master")) {
@@ -595,20 +702,12 @@ void SQLite3_Server_session_handler(MySQL_Session *sess, void *_pa, PtrSize_t *p
 #ifdef TEST_GROUPREP
 			if (strstr(query_no_space,(char *)"GR_MEMBER_ROUTING_CANDIDATE_STATUS")) {
 				pthread_mutex_unlock(&GloSQLite3Server->grouprep_mutex);
-				if (resultset->rows_count == 0) {
-					PROXY_TRACE();
-				}
 
-				if (strncmp("127.2.1.2", sess->client_myds->proxy_addr.addr,9) == 0) {
-					if (testTimeoutSequence[testIndex--])
-						sleep(2);
-					if (testIndex < 0)
-						testIndex = 7;
-				}
-				else {
-					if (rand() % 20 == 0)
-						sleep(2);
-				}
+				// NOTE: Enable this just in case of manual testing
+				// if (rand() % 100 == 0) {
+				// 	// randomly add some latency on 1% of the traffic
+				// 	sleep(2);
+				// }
 			}
 #endif // TEST_GROUPREP
 			if (strstr(query_no_space,(char *)"Seconds_Behind_Master")) {
@@ -631,6 +730,18 @@ void SQLite3_Server_session_handler(MySQL_Session *sess, void *_pa, PtrSize_t *p
 	l_free(query_length,query);
 }
 
+#ifdef TEST_GROUPREP
+group_rep_status SQLite3_Server::grouprep_test_value(const std::string& srv_addr) {
+	group_rep_status cur_srv_st { "YES", "YES", 0 };
+
+	auto it = grouprep_map.find(srv_addr);
+	if (it != grouprep_map.end()) {
+		cur_srv_st = it->second;
+	}
+
+	return cur_srv_st;
+}
+#endif
 
 SQLite3_Session::SQLite3_Session() {
 	sessdb=new SQLite3DB();
@@ -705,6 +816,22 @@ static void *child_mysql(void *arg) {
 		if (myds->net_failure) goto __exit_child_mysql;
 		myds->read_pkts();
 		sess->to_process=1;
+
+		// Get and set the client address before the sesion is processed.
+		union {
+			struct sockaddr_in in;
+			struct sockaddr_in6 in6;
+		} custom_sockaddr;
+		struct sockaddr *addr=(struct sockaddr *)malloc(sizeof(custom_sockaddr));
+		socklen_t addrlen=sizeof(custom_sockaddr);
+		memset(addr, 0, sizeof(custom_sockaddr));
+		sess->client_myds->client_addrlen=addrlen;
+		sess->client_myds->client_addr=addr;
+		int g_rc = getpeername(sess->client_myds->fd, addr, &addrlen);
+		if (g_rc == -1) {
+			proxy_error("'getpeername' failed with error: %d\n", rc);
+		}
+
 		int rc=sess->handler();
 		if (rc==-1) goto __exit_child_mysql;
 	}
@@ -912,7 +1039,16 @@ void SQLite3_Server::init_grouprep_ifaces_string(std::string& s) {
 	pthread_mutex_init(&grouprep_mutex,NULL);
 	if (!s.empty())
 		s += ";";
-	s += "127.2.1.1:3306;127.2.1.2:3306;127.2.1.3:3306";
+
+	// Maximum number of servers to simulate.
+	max_num_grouprep_servers = 50;
+	for (unsigned int i=0; i < max_num_grouprep_servers; i++) {
+		s += "127.2.1." + std::to_string(i) + ":3306";
+
+		if (i != max_num_grouprep_servers) {
+			s += ";";
+		}
+	}
 }
 #endif // TEST_GROUPREP
 
@@ -1094,19 +1230,77 @@ void SQLite3_Server::populate_aws_aurora_table(MySQL_Session *sess) {
 #endif // TEST_AURORA
 
 #ifdef TEST_GROUPREP
+/**
+ * @brief Populates the 'grouprep' table if it's found empty with the default
+ *   values for the three testing servers.
+ *
+ *   NOTE: This function needs to be called with lock on grouprep_mutex already acquired
+ *
+ * @param sess The current session performing a query.
+ * @param txs_behind Unused parameter.
+ */
 void SQLite3_Server::populate_grouprep_table(MySQL_Session *sess, int txs_behind) {
-	// this function needs to be called with lock on mutex galera_mutex already acquired
+	GloAdmin->mysql_servers_wrlock();
+	// We are going to repopulate the map
+	this->grouprep_map.clear();
+
+	char *error=NULL;
+	int cols=0;
+	int affected_rows=0;
+	SQLite3_result *resultset=NULL;
+
+	string query { "SELECT * FROM GR_MEMBER_ROUTING_CANDIDATE_STATUS" };
+	sessdb->execute_statement(query.c_str(), &error, &cols, &affected_rows, &resultset);
+	if (resultset) {
+		for (const SQLite3_row* r : resultset->rows) {
+			std::string srv_addr { std::string(r->fields[0]) + ":" + std::string(r->fields[1]) };
+			const group_rep_status srv_status {
+				std::string { r->fields[2] } == "YES" ? true : false,
+				std::string { r->fields[3] } == "YES" ? true : false,
+				atoi(r->fields[4])
+			};
+
+			this->grouprep_map[srv_addr] = srv_status;
+		}
+	}
+	delete resultset;
+
+	// Insert some default servers for manual testing.
 	//
-	sessdb->execute("DELETE FROM GR_MEMBER_ROUTING_CANDIDATE_STATUS");
-	string myip = string(sess->client_myds->proxy_addr.addr);
-	string server_id = myip.substr(8,1);
-	if (server_id == "1")
-		sessdb->execute("INSERT INTO GR_MEMBER_ROUTING_CANDIDATE_STATUS (viable_candidate, read_only, transactions_behind) values ('YES', 'NO', 0)");
-	else {
-		std::stringstream ss;
-		ss << "INSERT INTO GR_MEMBER_ROUTING_CANDIDATE_STATUS (viable_candidate, read_only, transactions_behind) values ('YES', 'YES', " << txs_behind << ")";
-		sessdb->execute(ss.str().c_str());
+	// NOTE: This logic can be improved in the future, for now it only populates
+	// the 'monitoring' data for the default severs. If more servers are placed
+	// as the default ones, more servers will be placed in their appropiated
+	// hostgroups with the same pattern as first ones.
+	if (this->grouprep_map.size() == 0) {
+		GloAdmin->admindb->execute_statement(
+			(char*)"SELECT DISTINCT hostname, port, hostgroup_id FROM mysql_servers"
+			" WHERE hostgroup_id BETWEEN 2700 AND 4200",
+			&error, &cols , &affected_rows , &resultset
+		);
+
+		for (const SQLite3_row* r : resultset->rows) {
+			std::string hostname { r->fields[0] };
+			int port = atoi(r->fields[1]);
+			int hostgroup_id = atoi(r->fields[2]);
+			const std::string t_insert_query {
+				"INSERT INTO GR_MEMBER_ROUTING_CANDIDATE_STATUS"
+					" (hostname, port, viable_candidate, read_only, transactions_behind) VALUES"
+					" ('%s', %d, '%s', '%s', 0)"
+			};
+			std::string insert_query {};
+
+			if (hostgroup_id % 4 == 0) {
+				string_format(t_insert_query, insert_query, hostname.c_str(), port, "YES", "NO");
+				sessdb->execute(insert_query.c_str());
+			} else {
+				string_format(t_insert_query, insert_query, hostname.c_str(), port, "YES", "YES");
+				sessdb->execute(insert_query.c_str());
+			}
+		}
+		delete resultset;
 	}
+
+	GloAdmin->mysql_servers_wrunlock();
 }
 #endif // TEST_GALERA
 
@@ -1179,7 +1373,11 @@ bool SQLite3_Server::init() {
 	tables_defs_grouprep = new std::vector<table_def_t *>;
 	insert_into_tables_defs(tables_defs_grouprep,
 		(const char *)"GR_MEMBER_ROUTING_CANDIDATE_STATUS",
-		(const char*)"CREATE TABLE GR_MEMBER_ROUTING_CANDIDATE_STATUS (viable_candidate varchar not null, read_only varchar not null, transactions_behind int not null)");
+		(const char*)"CREATE TABLE GR_MEMBER_ROUTING_CANDIDATE_STATUS ("
+			"hostname VARCHAR NOT NULL, port INT NOT NULL, viable_candidate varchar not null, read_only varchar not null, transactions_behind int not null, PRIMARY KEY (hostname, port)"
+		")"
+	);
+
 	check_and_build_standard_tables(sessdb, tables_defs_grouprep);
 	GloAdmin->enable_grouprep_testing();
 #endif // TEST_GALERA


Test patch:
diff --git a/test/tap/tests/client_host_err/create_netns_n.sh b/test/tap/tests/client_host_err/create_netns_n.sh
new file mode 100755
index 0000000000..4128030d6a
--- /dev/null
+++ b/test/tap/tests/client_host_err/create_netns_n.sh
@@ -0,0 +1,25 @@
+#!/bin/bash
+if [[ $EUID -ne 0 ]]; then
+   echo "This script must be run as root"
+   exit 1
+fi
+
+if [ ! "$1" ] || [ ! -z "${1//[0-9]}" ]; then
+    echo "Parameter should be a number"
+    exit 1
+fi
+
+ip netns add ns$1
+
+ip netns exec ns$1 ip link set lo up
+ip netns exec ns$1 ip link set dev lo up
+
+ip link add v-eth$1 type veth peer name v-peer1
+ip link set v-peer1 netns ns$1
+ip netns exec ns$1 ip addr
+ip netns exec ns$1 ip addr add 10.200.$1.2/24 dev v-peer1
+ip netns exec ns$1 ip link set v-peer1 up
+ip addr add 10.200.$1.1/24 dev v-eth$1
+ip link set v-eth$1 up
+ip netns exec ns$1 ip route add default via 10.200.$1.1
+
diff --git a/test/tap/tests/client_host_err/delete_netns_n.sh b/test/tap/tests/client_host_err/delete_netns_n.sh
new file mode 100755
index 0000000000..80dfbcc5fc
--- /dev/null
+++ b/test/tap/tests/client_host_err/delete_netns_n.sh
@@ -0,0 +1,14 @@
+#!/bin/bash
+if [[ $EUID -ne 0 ]]; then
+   echo "This script must be run as root"
+   exit 1
+fi
+
+if [ ! "$1" ] || [ ! -z "${1//[0-9]}" ]; then
+    echo "Parameter should be a number"
+    exit 1
+fi
+
+ip netns delete ns$1
+ip link delete v-eth$1
+
diff --git a/test/tap/tests/reg_test_3603-stmt_metadata-t.cpp b/test/tap/tests/reg_test_3603-stmt_metadata-t.cpp
index c47c3cf129..e116142da4 100644
--- a/test/tap/tests/reg_test_3603-stmt_metadata-t.cpp
+++ b/test/tap/tests/reg_test_3603-stmt_metadata-t.cpp
@@ -1,3 +1,26 @@
+/**
+ * @file reg_test_3603-stmt_metadata-t.cpp
+ * @brief This test is a regression test for issue #3603. It performs several
+ *   prepared statements using NULL values to check that these values are correctly
+ *   preserved/forgotten when new values are received.
+ * @details For being able to trigger the issue, the test needs to execute the
+ *   statements using 'MYSQL_TYPE_NULL' for supplied 'buffer_types'. Since the
+ *   issue consists in the invalid preservation of the 'NULL' values in
+ *   subsequent executions of the same prepared stmt. This tests performs two
+ *   sequences of actions:
+ *
+ *   1. Sets all the values to 'NULL' in a particular stmt, for later updating
+ *   those values in the subsequent execution of the stmt. If the issue is
+ *   present, reported values should be 'NULL' due to the issue.
+ *   2. Sets all the values of the same stmt to other values than NULL. Because
+ *   of the issue, this stmt needs to be executed twice to if the values were
+ *   'nullified' previously. Later executes an statement making this values to
+ *   NULL, followed by one with values other than NULL. This creates the
+ *   required conditions to provoque a 'heap-buffer-overflow'.
+ *
+ * @date 2020-09-07
+ */
+
 #include <vector>
 #include <string>
 #include <stdio.h>
diff --git a/test/tap/tests/reg_test_3625-sqlite3_session_client_error_limit-t.cpp b/test/tap/tests/reg_test_3625-sqlite3_session_client_error_limit-t.cpp
new file mode 100644
index 0000000000..a56669236d
--- /dev/null
+++ b/test/tap/tests/reg_test_3625-sqlite3_session_client_error_limit-t.cpp
@@ -0,0 +1,243 @@
+/**
+ * @file reg_test_3625-sqlite3_session_client_error_limit-t.cpp
+ * @brief Regression test check that performing a valid connection to ProxySQL SQLite3 server
+ *   should be successful when 'client_error_limit' is enabled.
+ * @details It performs the following operations:
+ *  - Enabling client error limit feature.
+ *  - Perform a connection to ProxySQL SQLite3 server.
+ */
+
+#include <cstring>
+#include <vector>
+#include <tuple>
+#include <string>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+
+#include <mysql.h>
+#include <mysql/mysqld_error.h>
+
+#include "tap.h"
+#include "command_line.h"
+#include "utils.h"
+
+using query_spec = std::tuple<std::string, int>;
+
+const int sqlite3_port = 0;
+
+/**
+ * @brief Extract the current 'sqliteserver-mysql_ifaces' from ProxySQL config.
+ * @param proxysql_admin An already opened connection to ProxySQL Admin.
+ * @return EXIT_SUCCESS, or one of the following error codes:
+ *   - EINVAL if supplied 'proxysql_admin' is NULL.
+ *   - '-1' in case of ProxySQL returns an 'NULL' row for the query selecting
+ *     the variable 'sqliteserver-read_only'.
+ *   - EXIT_FAILURE in case other operation failed.
+ */
+int get_sqlite3_ifaces(MYSQL* proxysql_admin, std::string& sqlite3_ifaces) {
+	if (proxysql_admin == NULL) {
+		return EINVAL;
+	}
+
+	int res = EXIT_FAILURE;
+
+	MYSQL_QUERY(
+		proxysql_admin,
+		"SELECT * FROM global_variables WHERE Variable_name='sqliteserver-mysql_ifaces'"
+	);
+
+	MYSQL_RES* admin_res = mysql_store_result(proxysql_admin);
+	if (!admin_res) {
+		diag("'mysql_store_result' at line %d failed: %s", __LINE__, mysql_error(proxysql_admin));
+		goto cleanup;
+	}
+
+	{
+		MYSQL_ROW row = mysql_fetch_row(admin_res);
+		if (!row || row[0] == nullptr || row[1] == nullptr) {
+			diag("'mysql_fetch_row' at line %d returned 'NULL'", __LINE__);
+			res = -1;
+			goto cleanup;
+		}
+
+		std::string _sqlite3_ifaces { row[1] };
+		sqlite3_ifaces = _sqlite3_ifaces;
+		res = EXIT_SUCCESS;
+	}
+
+cleanup:
+
+	return res;
+}
+
+int extract_sqlite3_host_port(MYSQL* proxysql_admin, std::pair<std::string, int>& host_port) {
+	if (proxysql_admin == nullptr) { return EINVAL; }
+	int res = EXIT_SUCCESS;
+
+	std::string sqlite3_ifaces {};
+	int ifaces_err = get_sqlite3_ifaces(proxysql_admin, sqlite3_ifaces);
+
+	// ProxySQL is likely to have been launched without "--sqlite3-server" flag
+	if (ifaces_err == -1) {
+		diag("ProxySQL was launched without '--sqlite3-server' flag");
+		res = EXIT_FAILURE;
+		return res;
+	}
+
+	// Extract the correct port to connect to SQLite server
+	std::string::size_type colon_pos = sqlite3_ifaces.find(":");
+	if (colon_pos == std::string::npos) {
+		diag("ProxySQL returned a malformed 'sqliteserver-mysql_ifaces': %s", sqlite3_ifaces.c_str());
+		res = EXIT_FAILURE;
+		return res;
+	}
+
+	std::string sqlite3_host { sqlite3_ifaces.substr(0, colon_pos) };
+	std::string sqlite3_port { sqlite3_ifaces.substr(colon_pos + 1) };
+
+	// Check that port has valid conversion
+	char* end_pos = nullptr;
+	int i_sqlite3_port = std::strtol(sqlite3_port.c_str(), &end_pos, 10);
+
+	if (errno == ERANGE || (end_pos != &sqlite3_port.back() + 1)) {
+		diag(
+			"ProxySQL returned a invalid port number within 'sqliteserver-mysql_ifaces': %s",
+			sqlite3_ifaces.c_str()
+		);
+		res = EXIT_FAILURE;
+		return res;
+	}
+
+	if (res == EXIT_SUCCESS) {
+		host_port = { sqlite3_host, i_sqlite3_port };
+	}
+
+	return res;
+}
+
+int main(int argc, char** argv) {
+	CommandLine cl;
+
+	// plan as many tests as queries
+	plan(
+		/* Enabling 'client_error_limit' succeed */
+		2 +
+		/* Fail to connect with wrong username and password */ 
+		2 +
+		/* Correctly connects with proper username and password */
+		1
+	);
+
+	if (cl.getEnv()) {
+		diag("Failed to get the required environmental variables.");
+		return EXIT_FAILURE;
+	}
+
+	MYSQL* proxysql_admin = mysql_init(NULL);
+
+	// Connect to ProxySQL Admin and check current SQLite3 configuration
+	if (
+		!mysql_real_connect(
+			proxysql_admin, cl.host, cl.admin_username, cl.admin_password,
+			NULL, cl.admin_port, NULL, 0
+		)
+	) {
+		fprintf(
+			stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__,
+			mysql_error(proxysql_admin)
+		);
+		return EXIT_FAILURE;
+	}
+
+	{
+		std::pair<std::string, int> host_port {};
+		int host_port_err = extract_sqlite3_host_port(proxysql_admin, host_port); 
+		if (host_port_err) {
+			diag("Failed to get and parse 'sqliteserver-mysql_ifaces' at line '%d'", __LINE__);
+			goto cleanup;
+		}
+
+		MYSQL* proxysql_sqlite3 = mysql_init(NULL);
+
+		// Enable 'client_error_limit'
+		{
+			int err_limit_errno = mysql_query(proxysql_admin, "SET mysql-client_host_cache_size=10");
+			ok(err_limit_errno == EXIT_SUCCESS, "Successfully updated 'mysql-client_host_cache_size'");
+			err_limit_errno = mysql_query(proxysql_admin, "LOAD MYSQL VARIABLES TO RUNTIME");
+			ok(err_limit_errno == EXIT_SUCCESS, "Enabled 'client_host_cache_size'");
+		}
+
+		// Perform the invalid connections
+		{
+			// Connect with invalid username
+			std::string inv_user_err {};
+			bool failed_to_connect = false;
+			if (
+				!mysql_real_connect(
+					proxysql_sqlite3, host_port.first.c_str(), "foobar_user", cl.password,
+					NULL, host_port.second, NULL, 0
+				)
+			) {
+				inv_user_err = mysql_error(proxysql_sqlite3);
+				failed_to_connect = true;
+			}
+
+			ok(
+				failed_to_connect,
+				"An invalid user should fail to connect to SQLite3 server, error was: %s",
+				inv_user_err.c_str()
+			);
+
+			// Reinitialize MYSQL handle
+			mysql_close(proxysql_sqlite3);
+			proxysql_sqlite3 = mysql_init(NULL);
+
+			// Connect with invalid password
+			std::string inv_pass_err {};
+			failed_to_connect = false;
+			if (
+				!mysql_real_connect(
+					proxysql_sqlite3, host_port.first.c_str(), cl.username, "foobar_pass",
+					NULL, host_port.second, NULL, 0
+				)
+			) {
+				inv_pass_err = mysql_error(proxysql_sqlite3);
+				failed_to_connect = true;
+			}
+
+			ok(
+				failed_to_connect,
+				"An invalid user should fail to connect to SQLite3 server, error was: %s",
+				inv_pass_err.c_str()
+			);
+
+			// Reinitialize MYSQL handle
+			mysql_close(proxysql_sqlite3);
+			proxysql_sqlite3 = mysql_init(NULL);
+		}
+
+		// Perform the valid connection
+		{
+			// Correctly connect to SQLite3 server
+			MYSQL* connect_errno =
+				mysql_real_connect(
+					proxysql_sqlite3, host_port.first.c_str(), cl.username, cl.password,
+					NULL, host_port.second, NULL, 0
+				);
+
+			ok(
+				connect_errno != NULL,
+				"Connection should succeed when using a valid 'username:password' (%s:%s)",
+				cl.username, cl.password
+			);
+			mysql_close(proxysql_sqlite3);
+		}
+	}
+
+cleanup:
+
+	mysql_close(proxysql_admin);
+
+	return exit_status();
+}
diff --git a/test/tap/tests/test_client_limit_error.cpp b/test/tap/tests/test_client_limit_error.cpp
new file mode 100644
index 0000000000..ace8bdd76d
--- /dev/null
+++ b/test/tap/tests/test_client_limit_error.cpp
@@ -0,0 +1,776 @@
+/**
+ * @file test_client_limit_error.cpp
+ * @brief This test aims to verify the logic for the 'client_limit_error' feature.
+ *   NOTE: This test isn't executed automatically because it requires elevated privileges
+ *   for being able to run the scripts 'create_netns_n' and 'delete_netns_n' that creates
+ *   and delete the networks namespaces required for it.
+ *
+ * @details Right now the test verifies the following cases:
+ *  1. Enable the feature and checks that the error count is incremented when a single
+ *     client tries to connect and that the cache entry values match expected ones.
+ *  2. Flush the entries, and check that counting is performed properly for a
+ *     single cache entry.
+ *  3. Flush the entries, and check that counting is performed properly for
+ *     multiple cache entries.
+ *  4. Flush the entries, and check:
+ *       1. That counting is performed properly for multiple cache entries.
+ *       2. Connections fail after the limit for one client.
+ *       3. Clients are deleted after a succesfull connection is performed.
+ *  5. Flush the entries, fill the cache and check that when the
+ *  'mysql-client_host_error_counts' is changed at runtime, connections are denied
+ *  to a client exceeding the new limit.
+ *  6. Flush the entries, fill the cache and check that the when the
+ *  'mysql-client_host_cache_size' is changed at runtime:
+ *    1. The exceeding elements are cleaned with each new connection.
+ *    2. Check that is not relevant if the element was or not present 
+ *       in the cache.
+ *    3. Checks that a proper connection is performed succesfully and the element is
+ *    removed from the cache
+ *
+ * @date 2021-09-10
+ */
+
+#include <unistd.h>
+#include <signal.h>
+#include <pthread.h>
+#include <stdio.h>
+
+#include <atomic>
+#include <algorithm>
+#include <tuple>
+#include <vector>
+#include <string>
+#include <thread>
+#include <iostream>
+#include <fstream>
+
+#include <libconfig.h>
+#include <proxysql_utils.h>
+#include <mysql.h>
+
+#include "tap.h"
+#include "command_line.h"
+#include "utils.h"
+
+const uint32_t NUM_NETWORK_NAMESPACES = 5;
+
+using host_cache_entry = std::tuple<std::string, uint32_t, uint64_t>;
+
+inline unsigned long long monotonic_time() {
+  struct timespec ts;
+  clock_gettime(CLOCK_MONOTONIC, &ts);
+  return (((unsigned long long) ts.tv_sec) * 1000000) + (ts.tv_nsec / 1000);
+}
+
+std::vector<host_cache_entry> get_client_host_cache_entries(MYSQL* proxysql_admin) {
+	int rc = mysql_query(
+		proxysql_admin,
+		"SELECT * FROM stats.stats_mysql_client_host_cache ORDER BY client_address"
+	);
+
+	MYSQL_ROW row = NULL;
+	MYSQL_RES* proxy_res = mysql_store_result(proxysql_admin);
+	std::vector<host_cache_entry> host_cache_entries {};
+
+	while ((row = mysql_fetch_row(proxy_res))) {
+		std::string client_address = row[0];
+		uint32_t error_count = atoi(row[1]);
+		uint64_t last_update = atoll(row[2]);
+
+		host_cache_entries.push_back({client_address, error_count, last_update});
+	}
+
+	mysql_free_result(proxy_res);
+
+	return host_cache_entries;
+}
+
+int main(int, char**) {
+	int res = 0;
+	CommandLine cl;
+
+	if (cl.getEnv()) {
+		diag("Failed to get the required environmental variables.");
+		return -1;
+	}
+
+	plan(27);
+
+	MYSQL* proxysql_admin = mysql_init(NULL);
+
+	// Initialize connections
+	if (!proxysql_admin) {
+		fprintf(
+			stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__,
+			mysql_error(proxysql_admin)
+		);
+		return -1;
+	}
+
+	// Connect to ProxySQL Admin
+	if (
+		!mysql_real_connect(
+			proxysql_admin, cl.host, cl.admin_username, cl.admin_password,
+			NULL, cl.admin_port, NULL, 0
+		)
+	) {
+		fprintf(
+			stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__,
+			mysql_error(proxysql_admin)
+		);
+		return -1;
+	}
+
+	// Setup the virtual namespaces to be used by the test
+	diag("                Setting up testing network namespaces                ");
+	diag("*********************************************************************");
+	printf("\n");
+
+	int setup_ns_i = 0;
+	const std::string t_create_ns_command =
+		std::string { cl.workdir } + "/client_host_err/create_netns_n.sh %d";
+	for (setup_ns_i = 1; setup_ns_i < NUM_NETWORK_NAMESPACES; setup_ns_i++) {
+		std::string create_ns_command {};
+		string_format(t_create_ns_command, create_ns_command, setup_ns_i);
+
+		int c_err = system(create_ns_command.c_str());
+		if (c_err) {
+			diag(
+				"Failed to create netns number '%d' with err: '%d'",
+				setup_ns_i, c_err
+			);
+			goto cleanup;
+		}
+	}
+
+	printf("\n");
+	diag("*********************************************************************");
+	printf("\n");
+
+	diag("                 Performing queries and checks                       ");
+	diag("*********************************************************************");
+	printf("\n");
+
+	{
+		const std::string t_inv_user_command =
+			"ip netns exec ns%d mysql -h10.200.%d.1 -uinv_user -pinv_pass -P6033";
+		const std::string t_valid_connection_command {
+			"ip netns exec ns%d mysql -h10.200.%d.1 -uroot -proot -P6033 -e'DO 1' 2>&1"
+		};
+
+		// 1. Enable the feature check that error count is incremented when a
+		// new client fails to connect, and that the cache entry values are the
+		// expected ones.
+		{
+			printf("\n");
+			diag("                 START TEST NUMBER 1                         ");
+			diag("-------------------------------------------------------------");
+
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_cache_size=1");
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_error_counts=5");
+			MYSQL_QUERY(proxysql_admin, "LOAD MYSQL VARIABLES TO RUNTIME");
+
+			// There shouldn't be any other entries in the cache for this test.
+			MYSQL_QUERY(proxysql_admin, "PROXYSQL FLUSH MYSQL CLIENT HOSTS");
+
+			std::string inv_user_command {};
+			string_format(t_inv_user_command, inv_user_command, 1, 1);
+			uint64_t pre_command_time = monotonic_time();
+
+			diag("Performing connections to fill 'client_host_cache'");
+
+			printf("\n");
+			int inv_user_errno = system(inv_user_command.c_str());
+			diag("Client connection failed with error: %d", inv_user_errno);
+			printf("\n");
+
+			diag("Performing checks over 'client_host_cache'");
+
+			std::vector<host_cache_entry> entries =
+				get_client_host_cache_entries(proxysql_admin);
+
+			ok(
+				entries.size() == 1,
+				"'client_host_cache' entries should be '1' after issuing 'PROXYSQL FLUSH"
+				" MYSQL CLIENT HOSTS' and one failed connection."
+			);
+
+			if (entries.size() == 1) {
+				host_cache_entry unique_entry { entries.back() };
+				const std::string client_addr { std::get<0>(unique_entry) };
+				const uint32_t error_count { std::get<1>(unique_entry) };
+				const uint64_t last_updated { std::get<2>(unique_entry) };
+				uint64_t post_command_time = monotonic_time();
+
+				ok(
+					client_addr == "10.200.1.2" && error_count == 1 &&
+					(pre_command_time < last_updated < post_command_time),
+					"Entry should match expected values - exp(addr: %s, err_count: %d, last_updated: %ld < %ld < %ld),"
+					" act(addr: %s, err_count: %d, last_updated: %ld < %ld < %ld)",
+					"10.200.1.2", 1, pre_command_time, last_updated, post_command_time,
+					client_addr.c_str(), error_count, pre_command_time, last_updated, post_command_time
+				);
+			}
+		}
+
+		// 2. Flush the entries, and check that counting is performed properly for a
+		// single cache entry.
+		{
+			printf("\n");
+			diag("                 START TEST NUMBER 2                         ");
+			diag("-------------------------------------------------------------");
+
+
+			// There shouldn't be any other entries in the cache for this test.
+			MYSQL_QUERY(proxysql_admin, "PROXYSQL FLUSH MYSQL CLIENT HOSTS");
+			int errors = 0;
+
+			std::string inv_user_command {};
+			string_format(t_inv_user_command, inv_user_command, 1, 1);
+			uint64_t pre_command_time = monotonic_time();
+
+			printf("\n");
+			diag("Performing connections to fill 'client_host_cache'");
+			for (errors = 0; errors < 5; errors++) {
+				printf("\n");
+				int inv_user_errno = system(inv_user_command.c_str());
+				diag("Client connection failed with error: %d", inv_user_errno);
+			}
+			printf("\n");
+
+			diag("Performing checks over 'client_host_cache'");
+
+			std::vector<host_cache_entry> entries =
+				get_client_host_cache_entries(proxysql_admin);
+
+			ok(
+				entries.size() == 1,
+				"'client_host_cache' entries should be '1' after issuing 'PROXYSQL FLUSH"
+				" MYSQL CLIENT HOSTS' and one failed connection."
+			);
+
+			if (entries.size() == 1) {
+				host_cache_entry unique_entry { entries.back() };
+				const std::string client_addr { std::get<0>(unique_entry) };
+				const uint32_t error_count { std::get<1>(unique_entry) };
+				const uint64_t last_updated { std::get<2>(unique_entry) };
+				uint64_t post_command_time = monotonic_time();
+
+				ok(
+					client_addr == "10.200.1.2" && error_count == errors &&
+					(pre_command_time < last_updated < post_command_time),
+					"Entry should match expected values - exp(addr: %s, err_count: %d, last_updated: %ld < %ld < %ld),"
+					" act(addr: %s, err_count: %d, last_updated: %ld < %ld < %ld)",
+					"10.200.1.2", 1, pre_command_time, last_updated, post_command_time,
+					client_addr.c_str(), error_count, pre_command_time, last_updated, post_command_time
+				);
+			}
+		}
+
+		// 3. Flush the entries, and check that counting is performed properly for
+		// multiple cache entries.
+		{
+			printf("\n");
+			diag("                 START TEST NUMBER 3                         ");
+			diag("-------------------------------------------------------------");
+
+			// Increase cache size
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_cache_size=5");
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_error_counts=5");
+			MYSQL_QUERY(proxysql_admin, "LOAD MYSQL VARIABLES TO RUNTIME");
+
+			// There shouldn't be any other entries in the cache for this test.
+			MYSQL_QUERY(proxysql_admin, "PROXYSQL FLUSH MYSQL CLIENT HOSTS");
+			int errors = 0;
+
+			// Prepare several commands from different network namespaces
+			std::vector<std::string> inv_connection_commands {};
+			for (int i = 1; i < NUM_NETWORK_NAMESPACES; i++) {
+				std::string inv_user_command {};
+				string_format(t_inv_user_command, inv_user_command, i, i);
+				inv_connection_commands.push_back(inv_user_command);
+			}
+
+			uint64_t pre_command_time = monotonic_time();
+
+			printf("\n");
+			diag("Performing connections to fill 'client_host_cache'");
+			for (const auto inv_conn_command : inv_connection_commands) {
+				for (errors = 0; errors < 2; errors++) {
+					printf("\n");
+					int inv_user_errno = system(inv_conn_command.c_str());
+					diag("Client connection failed with error: %d", inv_user_errno);
+				}
+				printf("\n");
+			}
+
+			diag("Performing checks over 'client_host_cache'");
+
+			std::vector<host_cache_entry> entries =
+				get_client_host_cache_entries(proxysql_admin);
+
+			ok(
+				entries.size() == NUM_NETWORK_NAMESPACES - 1,
+				"'client_host_cache' entries should be 'NUM_NETWORK_NAMESPACES' after issuing 'PROXYSQL FLUSH"
+				" MYSQL CLIENT HOSTS' and 'NUM_NETWORK_NAMESPACES' failed connections. Entries: '%ld'",
+				entries.size()
+			);
+
+			if (entries.size() == NUM_NETWORK_NAMESPACES - 1) {
+				uint32_t entry_num = 1;
+
+				for (const auto& entry : entries) {
+					const std::string client_addr { std::get<0>(entry) };
+					const uint32_t error_count { std::get<1>(entry) };
+					const uint64_t last_updated { std::get<2>(entry) };
+					uint64_t post_command_time = monotonic_time();
+
+					std::string t_exp_client_addr { "10.200.%d.2" };
+					std::string exp_client_addr {};
+					string_format(t_exp_client_addr, exp_client_addr, entry_num);
+
+					ok(
+						client_addr == exp_client_addr && error_count == errors &&
+						(pre_command_time < last_updated < post_command_time),
+						"Entry should match expected values - exp(addr: %s, err_count: %d, last_updated: %ld < %ld < %ld),"
+						" act(addr: %s, err_count: %d, last_updated: %ld < %ld < %ld)",
+						exp_client_addr.c_str(), errors, pre_command_time, last_updated, post_command_time,
+						client_addr.c_str(), error_count, pre_command_time, last_updated, post_command_time
+					);
+
+					entry_num += 1;
+				}
+			}
+		}
+
+		// 4. Flush the entries, and check:
+		//   1. That counting is performed properly for multiple cache entries.
+		//   2. Connections fail after the limit for one client.
+		//   3. Clients are deleted after a succesfull connection is performed.
+		{
+			printf("\n");
+			diag("                 START TEST NUMBER 4                         ");
+			diag("-------------------------------------------------------------");
+
+			// Increase cache size
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_cache_size=5");
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_error_counts=5");
+			MYSQL_QUERY(proxysql_admin, "LOAD MYSQL VARIABLES TO RUNTIME");
+
+			// There shouldn't be any other entries in the cache for this test.
+			MYSQL_QUERY(proxysql_admin, "PROXYSQL FLUSH MYSQL CLIENT HOSTS");
+			int errors = 0;
+
+			// Prepare several commands from different network namespaces
+			std::vector<std::string> inv_connection_commands {};
+			for (int i = 1; i < NUM_NETWORK_NAMESPACES; i++) {
+				std::string inv_user_command {};
+				string_format(t_inv_user_command, inv_user_command, i, i);
+				inv_connection_commands.push_back(inv_user_command);
+			}
+
+			uint64_t pre_command_time = monotonic_time();
+
+			printf("\n");
+			diag("Performing connections to fill 'client_host_cache'");
+			for (const auto inv_conn_command : inv_connection_commands) {
+				for (errors = 0; errors < 3; errors++) {
+					printf("\n");
+					int inv_user_errno = system(inv_conn_command.c_str());
+					diag("Client connection failed with error: %d", inv_user_errno);
+				}
+				printf("\n");
+			}
+
+			printf("\n");
+			diag("1. Check that counting is perfomred properly over multiple 'client_host_cache'");
+
+			std::vector<host_cache_entry> entries =
+				get_client_host_cache_entries(proxysql_admin);
+
+			ok(
+				entries.size() == NUM_NETWORK_NAMESPACES - 1,
+				"'client_host_cache' entries should be 'NUM_NETWORK_NAMESPACES' after issuing 'PROXYSQL FLUSH"
+				" MYSQL CLIENT HOSTS' and 'NUM_NETWORK_NAMESPACES' failed connections. Entries: '%ld'",
+				entries.size()
+			);
+
+			if (entries.size() == NUM_NETWORK_NAMESPACES - 1) {
+				uint32_t entry_num = 1;
+
+				for (const auto& entry : entries) {
+					const std::string client_addr { std::get<0>(entry) };
+					const uint32_t error_count { std::get<1>(entry) };
+					const uint64_t last_updated { std::get<2>(entry) };
+					uint64_t post_command_time = monotonic_time();
+
+					std::string t_exp_client_addr { "10.200.%d.2" };
+					std::string exp_client_addr {};
+					string_format(t_exp_client_addr, exp_client_addr, entry_num);
+
+					ok(
+						client_addr == exp_client_addr && error_count == errors &&
+						(pre_command_time < last_updated < post_command_time),
+						"Entry should match expected values - exp(addr: %s, err_count: %d, last_updated: %ld < %ld < %ld),"
+						" act(addr: %s, err_count: %d, last_updated: %ld < %ld < %ld)",
+						exp_client_addr.c_str(), errors, pre_command_time, last_updated, post_command_time,
+						client_addr.c_str(), error_count, pre_command_time, last_updated, post_command_time
+					);
+
+					entry_num += 1;
+				}
+			}
+
+			printf("\n");
+			diag("Performing connections to fill 'client_host_cache'");
+
+			uint32_t expected_ns = 4;
+			const std::string expected_address { "10.200.4.2" };
+			uint32_t expected_errors = 5;
+
+			int limits = errors;
+			std::string inv_user_command_limit {};
+			const std::string t_stderr_inv_user_command {
+				"ip netns exec ns%d mysql -h10.200.%d.1 -uinv_user -pinv_pass -P6033 2>&1"
+			};
+			string_format(t_stderr_inv_user_command, inv_user_command_limit, 4, 4);
+
+			std::string command_res {};
+			for (int limits = errors; limits < 5 + 1; limits++) {
+				printf("\n");
+				int inv_user_limit_err = exec(inv_user_command_limit.c_str(), command_res);
+				diag("Client connection failed with error: (%d, %s)", inv_user_limit_err, command_res.c_str());
+			}
+			printf("\n");
+
+			diag("2. Checking the connection is denied when the limit is reached.");
+
+			auto limit_error = command_res.find("ERROR 2013 (HY000)");
+			ok(
+				limit_error != std::string::npos,
+				"Last connection should fail with 'ERROR 2013', it exceeded the error limit. ErrMsg: '%s'",
+				command_res.c_str()
+			);
+
+			std::vector<host_cache_entry> new_entries {
+				get_client_host_cache_entries(proxysql_admin)
+			};
+
+			auto cache_entry =
+				std::find_if(
+					std::begin(new_entries),
+					std::end(new_entries),
+					[&] (const host_cache_entry& elem) -> bool {
+						return std::get<0>(elem) == expected_address;
+					}
+				);
+
+			bool found_exp_values = false;
+			std::string client_address {};
+			uint32_t found_errors = 0;
+
+			if (cache_entry != std::end(new_entries)) {
+				client_address = std::get<0>(*cache_entry);
+				found_errors = std::get<1>(*cache_entry);
+
+				found_exp_values =
+					std::get<0>(*cache_entry) == expected_address &&
+					std::get<1>(*cache_entry) == expected_errors;
+			}
+
+			ok(
+				found_exp_values,
+				"Entry should match expected values - exp(addr: %s, err_count: %d), act(addr: %s, err_count: %d)",
+				expected_address.c_str(), expected_errors, client_address.c_str(), found_errors
+			);
+
+			diag("3. Check that clients are deleted from the cache when the connections are succesfully performed");
+
+			for (int i = 1; i < NUM_NETWORK_NAMESPACES; i++) {
+				// This client as exceeded the max failures
+
+				std::string valid_connection_command {};
+				string_format(t_valid_connection_command, valid_connection_command, i, i);
+
+				// Client has exceeded maximum connections failure is expected
+				if (i == 4) {
+					std::string command_res {};
+					exec(valid_connection_command, command_res);
+					auto limit_error = command_res.find("ERROR 2013 (HY000)");
+
+					ok(
+						limit_error != std::string::npos,
+						"Connection should fail due to limit exceeded. ErrMsg: '%s'", command_res.c_str()
+					);
+				} else {
+					int command_err = system(valid_connection_command.c_str());
+					ok(
+						command_err == 0,
+						"Connection should succeed for clients which limit haven't been exceeded."
+					);
+				}
+			}
+
+			new_entries = get_client_host_cache_entries(proxysql_admin);
+			ok(
+				new_entries.size() == 1 &&
+				std::get<0>(new_entries.back()) == "10.200.4.2",
+				"Only client address exceeding the limit should remain in the cache -"
+				" exp('10.200.4.2'), act('%s')", std::get<0>(new_entries.back()).c_str()
+			);
+		}
+
+		// 5. Flush the entries, fill the cache and check that the when the
+		// 'mysql-client_host_error_counts' is changed at runtime, connections are denied
+		// to a client exceeding the new limit.
+		{
+			printf("\n");
+			diag("                 START TEST NUMBER 5                         ");
+			diag("-------------------------------------------------------------");
+
+			// Increase cache size
+			printf("\n");
+			diag("Setting the value of 'mysql-client_host_cache_size' to '5'");
+
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_cache_size=5");
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_error_counts=5");
+			MYSQL_QUERY(proxysql_admin, "LOAD MYSQL VARIABLES TO RUNTIME");
+
+			// There shouldn't be any other entries in the cache for this test.
+			MYSQL_QUERY(proxysql_admin, "PROXYSQL FLUSH MYSQL CLIENT HOSTS");
+			int errors = 0;
+
+			std::string inv_user_command {};
+			string_format(t_inv_user_command, inv_user_command, 1, 1);
+			uint64_t pre_command_time = monotonic_time();
+
+			diag("Performing connections to fill 'client_host_cache'");
+
+			for (int i = 0; i < 4; i++) {
+				printf("\n");
+				int inv_user_errno = system(inv_user_command.c_str());
+				diag("Client connection failed with error: %d", inv_user_errno);
+				printf("\n");
+			}
+
+			diag("Decreasing the value of 'mysql-client_host_error_counts' to '3'");
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_error_counts=3");
+			MYSQL_QUERY(proxysql_admin, "LOAD MYSQL VARIABLES TO RUNTIME");
+
+			{
+				printf("\n");
+				std::string valid_user_command {};
+				string_format(t_valid_connection_command, valid_user_command, 1, 1);
+
+				std::string command_res {};
+				int valid_user_err = exec(valid_user_command.c_str(), command_res);
+				diag("Client connection failed with error: (%d, %s)", valid_user_err, command_res.c_str());
+
+				auto limit_error = command_res.find("ERROR 2013 (HY000)");
+				ok(
+					limit_error != std::string::npos,
+					"Last connection should fail with 'ERROR 2013', it exceeded the error limit. ErrMsg: '%s'",
+					command_res.c_str()
+				);
+			}
+		}
+
+		// 6. Flush the entries, fill the cache and check that the when the
+		// 'mysql-client_host_cache_size' is changed at runtime, the exceeding
+		// elements are cleaned with each new connection. Without being relevant if was
+		// present or not in the cache.
+		{
+			printf("\n");
+			diag("                 START TEST NUMBER 6                         ");
+			diag("-------------------------------------------------------------");
+
+			// Increase cache size
+			printf("\n");
+			diag("Setting the value of 'mysql-client_host_cache_size' to '5'");
+
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_cache_size=5");
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_error_counts=5");
+			MYSQL_QUERY(proxysql_admin, "LOAD MYSQL VARIABLES TO RUNTIME");
+
+			// There shouldn't be any other entries in the cache for this test.
+			MYSQL_QUERY(proxysql_admin, "PROXYSQL FLUSH MYSQL CLIENT HOSTS");
+			int errors = 0;
+
+			// Fill the cache with the entries from all the created namespaces
+			{
+				std::vector<std::string> inv_connection_commands {};
+				for (int i = 1; i < NUM_NETWORK_NAMESPACES; i++) {
+					std::string inv_user_command {};
+					string_format(t_inv_user_command, inv_user_command, i, i);
+					inv_connection_commands.push_back(inv_user_command);
+				}
+
+				printf("\n");
+				diag("Performing connections to fill 'client_host_cache'");
+				for (const auto inv_conn_command : inv_connection_commands) {
+					for (errors = 0; errors < 3; errors++) {
+						printf("\n");
+						int inv_user_errno = system(inv_conn_command.c_str());
+						diag("Client connection failed with error: %d", inv_user_errno);
+					}
+					printf("\n");
+				}
+				printf("\n");
+			}
+
+			diag("Decreasing the value of 'mysql-client_host_cache_size' to '3'");
+			MYSQL_QUERY(proxysql_admin, "SET mysql-client_host_cache_size=1");
+			MYSQL_QUERY(proxysql_admin, "LOAD MYSQL VARIABLES TO RUNTIME");
+
+			// Update the latest entry in the cache, oldest member "10.200.1.2" should go away.
+			{
+				uint64_t pre_command_time = monotonic_time();
+
+				diag("1. Checking that the connection updates the entry and the oldest entry is removed");
+
+				std::string inv_user_command {};
+				string_format(t_inv_user_command, inv_user_command, 4, 4);
+
+				printf("\n");
+				int inv_user_err = system(inv_user_command.c_str());
+				diag("Client connection failed with error: %d", inv_user_err);
+				printf("\n");
+
+				std::vector<host_cache_entry> updated_entries {
+					get_client_host_cache_entries(proxysql_admin)
+				};
+
+				std::string exp_client_addr { "10.200.4.2" };
+
+				auto entry = std::find_if(
+					std::begin(updated_entries),
+					std::end(updated_entries),
+					[&] (const host_cache_entry& entry) -> bool {
+						return std::get<0>(entry) == exp_client_addr;
+					}
+				);
+
+				std::string act_client_addr {};
+				uint64_t last_updated = 0;
+
+				if (entry != std::end(updated_entries)) {
+					act_client_addr = std::get<0>(*entry);
+					last_updated = std::get<2>(*entry);
+				}
+
+				ok(
+					exp_client_addr == act_client_addr && last_updated > pre_command_time,
+					"Entry should be present and updated with the following values -"
+					" exp('%s', %ld > %ld), act('%s', %ld > %ld)", exp_client_addr.c_str(),
+					last_updated, pre_command_time, act_client_addr.c_str(), last_updated,
+					pre_command_time
+				);
+
+				// Oldest member shouldn't be present
+				std::string oldest_member { "10.200.1.2" };
+
+				auto oldest_entry = std::find_if(
+					std::begin(updated_entries),
+					std::end(updated_entries),
+					[&] (const host_cache_entry& entry) -> bool {
+						return std::get<0>(entry) == oldest_member;
+					}
+				);
+
+				ok(
+					oldest_entry == std::end(updated_entries),
+					"Oldest entry '%s' shouldn't be present in the cache.", oldest_member.c_str()
+				);
+
+				printf("\n");
+				diag("2. Checking that the same behavior is observed if connection comes from a non-cached address");
+
+				string_format(t_inv_user_command, inv_user_command, 1, 1);
+
+				printf("\n");
+				inv_user_err = system(inv_user_command.c_str());
+				diag("Client connection failed with error: %d", inv_user_err);
+				printf("\n");
+
+				diag("2.1 Checking that the address hasn't been added");
+
+				const std::string new_member { "10.200.1.2" };
+
+				updated_entries = get_client_host_cache_entries(proxysql_admin);
+				auto new_entry = std::find_if(
+					std::begin(updated_entries),
+					std::end(updated_entries),
+					[&] (const host_cache_entry& entry) -> bool {
+						return std::get<0>(entry) == new_member;
+					}
+				);
+
+				ok(
+					new_entry == std::end(updated_entries),
+					"New entry from address '10.200.1.2' shouldn't be present in the cache"
+				);
+
+				printf("\n");
+				diag("2.1 Checking that the oldest address has been removed");
+				oldest_member = "10.200.2.2";
+
+				oldest_entry = std::find_if(
+					std::begin(updated_entries),
+					std::end(updated_entries),
+					[&] (const host_cache_entry& entry) -> bool {
+						return std::get<0>(entry) == oldest_member;
+					}
+				);
+
+				ok(
+					oldest_entry == std::end(updated_entries),
+					"Oldest entry '%s' shouldn't be present in the cache.", oldest_member.c_str()
+				);
+
+				printf("\n");
+				diag("2.2 Checking that a successful connection gets a client removed");
+				printf("\n");
+
+				std::string valid_connection_command {};
+				string_format(t_valid_connection_command, valid_connection_command, 3, 3);
+				system(valid_connection_command.c_str());
+
+				const std::string forgotten_address { "10.200.3.2" };
+
+				updated_entries = get_client_host_cache_entries(proxysql_admin);
+				auto forgot_entry = std::find_if(
+					std::begin(updated_entries),
+					std::end(updated_entries),
+					[&] (const host_cache_entry& entry) -> bool {
+						return std::get<0>(entry) == forgotten_address;
+					}
+				);
+
+				ok(
+					forgot_entry == std::end(updated_entries),
+					"Entry '%s' should have been forgotten due to successful connection.",
+					forgotten_address.c_str()
+				);
+			}
+		}
+	}
+
+cleanup:
+	// Cleanup the virtual namespaces to be used by the test
+	printf("\n");
+	diag("            Cleanup of testing network namespaces                    ");
+	diag("*********************************************************************");
+	printf("\n");
+
+	const std::string t_delete_ns_command =
+		std::string { cl.workdir } + "/client_host_err/delete_netns_n.sh %d";
+	for (int i = 1; i < setup_ns_i; i++) {
+		std::string delete_ns_command {};
+		string_format(t_delete_ns_command, delete_ns_command, i);
+		system(delete_ns_command.c_str());
+	}
+
+	mysql_close(proxysql_admin);
+
+	return exit_status();
+}