Avoid 100% CPU usage while socket is closed

[orange-kao]

After stop/start kafka service, kafka-python may use 100% CPU caused by
busy-retry while the socket was closed. This fix the issue by unregister
the socket if the fd is negative.

---

This change is 

[ghost]

It could also be possible to do a self._selector.unregister(key.fileobj) instead, but the interesting part is where this negative fd is actually coming from, that might be the root cause for this issue, but might also be tricky to find.

[orange-kao]

Commit updated. Unregister socket instead of sleep.

[jeffwidman]

Agreed, can you look at where this negative FD might be coming from? If you can consistently repro it, then the eBPF tools would probably make it a lot easier to track down...

Or if you have a way to consistently repro it in a test case, I'd be willing to take a look...

[gabriel-tincu]

Agreed, can you look at where this negative FD might be coming from? If you can consistently repro it, then the eBPF tools would probably make it a lot easier to track down...

Or if you have a way to consistently repro it in a test case, I'd be willing to take a look...

@jeffwidman the scenario where this happens, to the best of my knowledge, is having the same client connected to a cluster that rapidly drops and accepts new members. Having a long-ish metadata refresh time, plus the client code, to my understanding, having a very long grace period to kill idle connections, can lead to the least_loaded_node method returning nodes that are no longer actually there, mapped to dead sockets. Needless to say, setting up a scenario for this would require some automation (1 broker leaves, another arrives, least_loaded_node is forced to select the old one or we just use that node id

[dpkp]

I think this needs to be more robust: we want to close the conn here if it is a BrokerConnection, which would then trigger an unregister. But it could also be the _wake_r socketpair, in which case we need to reset/rebuild the wake socketpair.

[dnj12345]

Agreed, can you look at where this negative FD might be coming from? If you can consistently repro it, then the eBPF tools would probably make it a lot easier to track down...
Or if you have a way to consistently repro it in a test case, I'd be willing to take a look...

@jeffwidman the scenario where this happens, to the best of my knowledge, is having the same client connected to a cluster that rapidly drops and accepts new members. Having a long-ish metadata refresh time, plus the client code, to my understanding, having a very long grace period to kill idle connections, can lead to the least_loaded_node method returning nodes that are no longer actually there, mapped to dead sockets. Needless to say, setting up a scenario for this would require some automation (1 broker leaves, another arrives, least_loaded_node is forced to select the old one or we just use that node id

I am also seeing this issue. When the broker goes down, the CPU usage percentage of the producer shoots up. In htop my producer usage is usually less than 5% under normal conditions. Once the broker goes down and CPU shoots up to 50% usage. There is a constant stream of records I am trying to write to the bus. The producer.send call does not raise an exception which would have helped me close the producer.