From 93e78d1f034e32406d83b3ed220562773b01c44e Mon Sep 17 00:00:00 2001
From: Jarrod Johnson <jjohnson2@lenovo.com>
Date: Thu, 28 Jun 2018 09:53:34 -0400
Subject: [PATCH] Fix try_assimilate

Also get rid of CLOSE_WAIT in some situations.
---
 confluent_server/confluent/collective/manager.py   | 3 +++
 confluent_server/confluent/config/configmanager.py | 1 +
 2 files changed, 4 insertions(+)

diff --git a/confluent_server/confluent/collective/manager.py b/confluent_server/confluent/collective/manager.py
index 19a817c6..d9f7bc54 100644
--- a/confluent_server/confluent/collective/manager.py
+++ b/confluent_server/confluent/collective/manager.py
@@ -96,6 +96,7 @@ def connect_to_leader(cert=None, name=None, leader=None):
 def follow_leader(remote):
     cfm.follow_channel(remote)
     # The leader has folded, time to startup again...
+    remote.close()
     eventlet.spawn_n(start_collective)
 
 
@@ -272,6 +273,8 @@ def try_assimilate(drone):
     tlvdata.send(remote, {'collective': {'operation': 'assimilate',
                                          'name': get_myname(),
                                          'txcount': cfm._txcount}})
+    tlvdata.recv(remote)  # the banner
+    tlvdata.recv(remote)  # authpassed... 0..
     answer = tlvdata.recv(remote)
     if 'error' in answer:
         connect_to_leader(None, None, leader=remote.getpeername()[0])
diff --git a/confluent_server/confluent/config/configmanager.py b/confluent_server/confluent/config/configmanager.py
index 7bcf4e49..eb5c1348 100644
--- a/confluent_server/confluent/config/configmanager.py
+++ b/confluent_server/confluent/config/configmanager.py
@@ -475,6 +475,7 @@ def relay_slaved_requests(name, listener):
             if 'xid' in rpc:
                 _push_rpc(listener, cPickle.dumps({'xid': rpc['xid']}))
         msg = listener.recv(8)
+    listener.close()
     del cfgstreams[name]
     if not cfgstreams:
         cfgleader = True