+ /*
+ * We lock here because we are about to close the sockets and some other
+ * thread might be using them so wait before we are exclusive which will
+ * abort all other consumer command by other threads.
+ */
+ pthread_mutex_lock(&consumer_data->lock);
+